Skip to content

API Reference

Twain

The central entry point for interacting with a TwainMQ installation.

A Twain instance represents a complete TwainMQ environment rooted at a directory on disk. This directory contains all topics, configuration files, and global metadata. Applications typically create one long-lived Twain object and use it to manage topics, producers, consumers, and message registrations.

Parameters

root_dir : str or Path Filesystem path to the TwainMQ root directory. This directory will hold all topic data, configuration, and state.

Message Classes

Dataclass-based messages must be registered with the Twain instance before they can be produced or consumed. Registration assigns each message type a stable numeric identifier used during encoding. All producers and consumers created from the same Twain share this registry.

In rare cases where different topics require dataclasses with the same message name but different schemas, separate Twain instances must be used. (It is strongly recommended to avoid such naming collisions.)

Notes

  • A Twain object is intended to be long-lived and reused across your application.
  • Global configuration (e.g. message type registry) is stored at the Twain level and applies to all producers and consumers created from it.
  • Topics created through this instance are immediately available for producing and consuming messages.

Examples

Create a Twain instance pointing at a local directory:

tmq = Twain("C:/TwainMQ")

Create a new topic with 16-bit unsigned integer keys:

tmq.create_topic("hello_world", "u16")

Produce a message:

producer = tmq.producer("hello_world") producer.write_message(42, "Hello!")

Consume messages:

consumer = tmq.consumer("hello_world") msg = consumer.poll()

Source code in src/twainmq/core.py 
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
class Twain:
    """
    The central entry point for interacting with a TwainMQ installation.

    A `Twain` instance represents a complete TwainMQ environment rooted at a
    directory on disk. This directory contains all topics, configuration files,
    and global metadata. Applications typically create one long-lived `Twain`
    object and use it to manage topics, producers, consumers, and message
    registrations.

    Parameters
    ----------
    root_dir : str or Path
        Filesystem path to the TwainMQ root directory. This directory will hold
        all topic data, configuration, and state.

    Message Classes
    ---------------
    Dataclass-based messages must be registered with the `Twain` instance before
    they can be produced or consumed. Registration assigns each message type a
    stable numeric identifier used during encoding. All producers and consumers
    created from the same `Twain` share this registry.

    In rare cases where different topics require dataclasses with the same
    message name but different schemas, separate `Twain` instances must be used.
    (It is strongly recommended to avoid such naming collisions.)

    Notes
    -----
    - A `Twain` object is intended to be long-lived and reused across your
      application.
    - Global configuration (e.g. message type registry) is stored at the `Twain`
      level and applies to all producers and consumers created from it.
    - Topics created through this instance are immediately available for
      producing and consuming messages.

    Examples
    --------
    Create a Twain instance pointing at a local directory:

    >>> tmq = Twain("C:/TwainMQ")

    Create a new topic with 16-bit unsigned integer keys:

    >>> tmq.create_topic("hello_world", "u16")

    Produce a message:

    >>> producer = tmq.producer("hello_world")
    >>> producer.write_message(42, "Hello!")

    Consume messages:

    >>> consumer = tmq.consumer("hello_world")
    >>> msg = consumer.poll()
    """
    def __init__(self, root_dir):
        self._root_dir = Path(root_dir)
        self._msg_cls_registry = dict()
        for m in CONSUMER_GROUP_MESSAGE_CLASSES:
            self.register_msg_cls(m)

    def register_msg_cls(self, message_cls):
        """
        Register a dataclass message type for use with this Twain instance.

        Dataclass messages must be registered before they can be produced or
        consumed. Registration assigns the class a stable numeric identifier used
        during encoding and decoding.

        Parameters
        ----------
        message_cls : type
            A dataclass defining a message schema. The class name (or its
            `__message_type__` attribute, if present) is used as the message type
            identifier.

        Raises
        ------
        KeyError
            If a message class with the same name is already registered.
        """
        name = getattr(message_cls, "__message_type__", message_cls.__name__)
        if name in self._msg_cls_registry:
            raise KeyError(f"Class already registered: {name}")
        self._msg_cls_registry[name] = message_cls

    def create_topic(self, topic_name, key_type=None, partitions=1, message_types=None):
        """
        Create a new topic in this TwainMQ instance.

        Parameters
        ----------
        topic_name : str
            Name of the topic. Must contain only safe characters.
        key_type : str, optional
            Key encoding type. Defaults to `"u16"`. Supported values are `"u8"`,
            `"u16"`, `"u32"`, `"u64"`, or `"charN"` where `N` is the number of
            characters.
        partitions : int, optional
            Number of partitions for the topic. Defaults to 1.
        message_types : list of str, optional
            Names of dataclass message types used by this topic. These do not need
            to be registered yet; they are recorded in the topic configuration.

        Raises
        ------
        InvalidTopicNameError
            If the topic name contains invalid characters.
        InvalidKeyTypeError
            If `key_type` is not recognised.
        ValueError
            If the topic already exists.
        """
        key_types = dict(
        u8 = 1,
        u16 = 2,
        u32 = 4,
        u64 = 8,
        )

        if not _is_safe(topic_name):
            raise InvalidTopicNameError("Topic name contains invalid characters")

        if key_type is None:
            key_type = "u16"

        try:
            if key_type.startswith("char") and key_type[4:].isdigit():
                key_width = -int(key_type[4:])
            else:
                key_width = key_types[key_type]
        except KeyError:
            raise InvalidKeyTypeError(f"{key_type} is not a valid key_type. Options are {', '.join(key_types.keys())} or charN where N is an integer")

        topic_path = self._topic_path(topic_name)
        if topic_path.exists():
            raise ValueError(f"Cannot create topic, {topic_name} already exists")
        new_topic_dir = topic_path.mkdir()
        config_path = self._config_path(topic_name)
        if message_types is None:
            message_types = {}
        else:
            message_types = {m: i for i, m in enumerate(message_types)}
        config = dict(
            key_width = key_width,
            partitions = partitions,
            message_types = message_types,
        )

        with config_path.open("w", encoding="utf-8") as f:
            json.dump(config, f, indent = 0)

    def delete_topic(self, topic_name):
        """
        Delete a topic and all of its data.

        This operation is irreversible. A confirmation prompt is shown to prevent
        accidental deletion.

        Parameters
        ----------
        topic_name : str
            The name of the topic to delete.

        Returns
        -------
        None or TopicDeleteError
            Returns an error object if the user fails the confirmation prompt.
        """
        challenge_digit = random.randint(0, 9)
        confirm = input(f"To confirm delete of topic {topic_name} in {self.root_dir}, type YES{challenge_digit}")
        if confirm == f"YES{challenge_digit}":
            shutil.rmtree(self._topic_path(topic_name))
            logger.info(f"Topic deleted: {topic_name}")
        else:
            return TopicDeleteError("User confirm failed, topic not deleted")

    def producer(self, topic_name):
        """
        Create a producer for the given topic.

        Parameters
        ----------
        topic_name : str
            The topic to produce messages to.

        Returns
        -------
        TwainMQProducer
            A producer bound to the specified topic.
        """
        return TwainMQProducer(self, topic_name)

    def consumer(self, topic_name, start_from=None, group=None):
        """
        Create a consumer for the given topic.

        Parameters
        ----------
        topic_name : str
            The topic to consume from.
        start_from : {"start", "now"}, optional
            Initial read position when no committed offset exists. Defaults to
            `"start"`.
        group : str or None, optional
            Consumer-group identifier. If provided, the consumer joins the group and
            participates in consensus-based partition assignment and commits.

        Returns
        -------
        TwainMQConsumer
            A consumer bound to the specified topic.
        """
        if start_from is None:
            start_from = "start"
        return TwainMQConsumer(self, topic_name, start_from, group)

    def _topic_path(self, topic_name: str) -> Path:
        return self.root_dir / topic_name

    def topic_exists(self, topic_name):
        """
        Check whether a topic exists in this TwainMQ instance.

        Parameters
        ----------
        topic_name : str

        Returns
        -------
        bool
            True if the topic exists, False otherwise.
        """
        return self._topic_path(topic_name).exists()

    def list_topics(self):
        """
        List all topics in this TwainMQ instance.

        Returns
        -------
        list of str
            Names of all topics, excluding internal consumer‑group directories.
        """
        return [t.stem for t in self.root_dir.iterdir() if not t.name.startswith("--group--") if t.name.endswith(".twc")]

    def _config_path(self, topic_name):
        return self.root_dir / f"{topic_name}.twc"

    @property
    def root_dir(self):
        return self._root_dir

consumer(topic_name, start_from=None, group=None)

Create a consumer for the given topic.

Parameters

topic_name : str The topic to consume from. start_from : {"start", "now"}, optional Initial read position when no committed offset exists. Defaults to "start". group : str or None, optional Consumer-group identifier. If provided, the consumer joins the group and participates in consensus-based partition assignment and commits.

Returns

TwainMQConsumer A consumer bound to the specified topic.

Source code in src/twainmq/core.py 
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
def consumer(self, topic_name, start_from=None, group=None):
    """
    Create a consumer for the given topic.

    Parameters
    ----------
    topic_name : str
        The topic to consume from.
    start_from : {"start", "now"}, optional
        Initial read position when no committed offset exists. Defaults to
        `"start"`.
    group : str or None, optional
        Consumer-group identifier. If provided, the consumer joins the group and
        participates in consensus-based partition assignment and commits.

    Returns
    -------
    TwainMQConsumer
        A consumer bound to the specified topic.
    """
    if start_from is None:
        start_from = "start"
    return TwainMQConsumer(self, topic_name, start_from, group)

create_topic(topic_name, key_type=None, partitions=1, message_types=None)

Create a new topic in this TwainMQ instance.

Parameters

topic_name : str Name of the topic. Must contain only safe characters. key_type : str, optional Key encoding type. Defaults to "u16". Supported values are "u8", "u16", "u32", "u64", or "charN" where N is the number of characters. partitions : int, optional Number of partitions for the topic. Defaults to 1. message_types : list of str, optional Names of dataclass message types used by this topic. These do not need to be registered yet; they are recorded in the topic configuration.

Raises

InvalidTopicNameError If the topic name contains invalid characters. InvalidKeyTypeError If key_type is not recognised. ValueError If the topic already exists.

Source code in src/twainmq/core.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
def create_topic(self, topic_name, key_type=None, partitions=1, message_types=None):
    """
    Create a new topic in this TwainMQ instance.

    Parameters
    ----------
    topic_name : str
        Name of the topic. Must contain only safe characters.
    key_type : str, optional
        Key encoding type. Defaults to `"u16"`. Supported values are `"u8"`,
        `"u16"`, `"u32"`, `"u64"`, or `"charN"` where `N` is the number of
        characters.
    partitions : int, optional
        Number of partitions for the topic. Defaults to 1.
    message_types : list of str, optional
        Names of dataclass message types used by this topic. These do not need
        to be registered yet; they are recorded in the topic configuration.

    Raises
    ------
    InvalidTopicNameError
        If the topic name contains invalid characters.
    InvalidKeyTypeError
        If `key_type` is not recognised.
    ValueError
        If the topic already exists.
    """
    key_types = dict(
    u8 = 1,
    u16 = 2,
    u32 = 4,
    u64 = 8,
    )

    if not _is_safe(topic_name):
        raise InvalidTopicNameError("Topic name contains invalid characters")

    if key_type is None:
        key_type = "u16"

    try:
        if key_type.startswith("char") and key_type[4:].isdigit():
            key_width = -int(key_type[4:])
        else:
            key_width = key_types[key_type]
    except KeyError:
        raise InvalidKeyTypeError(f"{key_type} is not a valid key_type. Options are {', '.join(key_types.keys())} or charN where N is an integer")

    topic_path = self._topic_path(topic_name)
    if topic_path.exists():
        raise ValueError(f"Cannot create topic, {topic_name} already exists")
    new_topic_dir = topic_path.mkdir()
    config_path = self._config_path(topic_name)
    if message_types is None:
        message_types = {}
    else:
        message_types = {m: i for i, m in enumerate(message_types)}
    config = dict(
        key_width = key_width,
        partitions = partitions,
        message_types = message_types,
    )

    with config_path.open("w", encoding="utf-8") as f:
        json.dump(config, f, indent = 0)

delete_topic(topic_name)

Delete a topic and all of its data.

This operation is irreversible. A confirmation prompt is shown to prevent accidental deletion.

Parameters

topic_name : str The name of the topic to delete.

Returns

None or TopicDeleteError Returns an error object if the user fails the confirmation prompt.

Source code in src/twainmq/core.py 
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
def delete_topic(self, topic_name):
    """
    Delete a topic and all of its data.

    This operation is irreversible. A confirmation prompt is shown to prevent
    accidental deletion.

    Parameters
    ----------
    topic_name : str
        The name of the topic to delete.

    Returns
    -------
    None or TopicDeleteError
        Returns an error object if the user fails the confirmation prompt.
    """
    challenge_digit = random.randint(0, 9)
    confirm = input(f"To confirm delete of topic {topic_name} in {self.root_dir}, type YES{challenge_digit}")
    if confirm == f"YES{challenge_digit}":
        shutil.rmtree(self._topic_path(topic_name))
        logger.info(f"Topic deleted: {topic_name}")
    else:
        return TopicDeleteError("User confirm failed, topic not deleted")

list_topics()

List all topics in this TwainMQ instance.

Returns

list of str Names of all topics, excluding internal consumer‑group directories.

Source code in src/twainmq/core.py 
251
252
253
254
255
256
257
258
259
260
def list_topics(self):
    """
    List all topics in this TwainMQ instance.

    Returns
    -------
    list of str
        Names of all topics, excluding internal consumer‑group directories.
    """
    return [t.stem for t in self.root_dir.iterdir() if not t.name.startswith("--group--") if t.name.endswith(".twc")]

producer(topic_name)

Create a producer for the given topic.

Parameters

topic_name : str The topic to produce messages to.

Returns

TwainMQProducer A producer bound to the specified topic.

Source code in src/twainmq/core.py 
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
def producer(self, topic_name):
    """
    Create a producer for the given topic.

    Parameters
    ----------
    topic_name : str
        The topic to produce messages to.

    Returns
    -------
    TwainMQProducer
        A producer bound to the specified topic.
    """
    return TwainMQProducer(self, topic_name)

register_msg_cls(message_cls)

Register a dataclass message type for use with this Twain instance.

Dataclass messages must be registered before they can be produced or consumed. Registration assigns the class a stable numeric identifier used during encoding and decoding.

Parameters

message_cls : type A dataclass defining a message schema. The class name (or its __message_type__ attribute, if present) is used as the message type identifier.

Raises

KeyError If a message class with the same name is already registered.

Source code in src/twainmq/core.py 
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def register_msg_cls(self, message_cls):
    """
    Register a dataclass message type for use with this Twain instance.

    Dataclass messages must be registered before they can be produced or
    consumed. Registration assigns the class a stable numeric identifier used
    during encoding and decoding.

    Parameters
    ----------
    message_cls : type
        A dataclass defining a message schema. The class name (or its
        `__message_type__` attribute, if present) is used as the message type
        identifier.

    Raises
    ------
    KeyError
        If a message class with the same name is already registered.
    """
    name = getattr(message_cls, "__message_type__", message_cls.__name__)
    if name in self._msg_cls_registry:
        raise KeyError(f"Class already registered: {name}")
    self._msg_cls_registry[name] = message_cls

topic_exists(topic_name)

Check whether a topic exists in this TwainMQ instance.

Parameters

topic_name : str

Returns

bool True if the topic exists, False otherwise.

Source code in src/twainmq/core.py 
236
237
238
239
240
241
242
243
244
245
246
247
248
249
def topic_exists(self, topic_name):
    """
    Check whether a topic exists in this TwainMQ instance.

    Parameters
    ----------
    topic_name : str

    Returns
    -------
    bool
        True if the topic exists, False otherwise.
    """
    return self._topic_path(topic_name).exists()

TwainMQProducer

Bases: TwainMQBase

A producer for writing messages to a TwainMQ topic.

Producers assign messages to partitions using a configurable partitioner and append them atomically to the topic's storage. Messages may be strings, bytes, or registered dataclass instances. Producers are lightweight and may be used directly or as context managers to ensure clean shutdown.

Source code in src/twainmq/producer.py 
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
class TwainMQProducer(TwainMQBase):
    """
    A producer for writing messages to a TwainMQ topic.

    Producers assign messages to partitions using a configurable partitioner and
    append them atomically to the topic's storage. Messages may be strings,
    bytes, or registered dataclass instances. Producers are lightweight and may
    be used directly or as context managers to ensure clean shutdown.
    """
    def __init__(self, twain: "Twain", topic: str, partitioner = None): 
        """
        Create a producer for a given topic.

        Parameters
        ----------
        twain : Twain
            The TwainMQ instance backing this producer.
        topic : str
            The topic to write messages to.
        partitioner : callable, optional
            A function mapping `(key, n_partitions)` to a partition index.
            If omitted, a stable 64-bit hash partitioner is used.
        """
        super().__init__(twain, topic)
        if partitioner is None:
            partitioner = partition_hash64
        self._partitioner = partitioner
        self._active_files, self._chunk_str = self._get_active_message_files()

    def __enter__(self):
        return self
    def __exit__(self, type, value, traceback):
        self.close()

    def encode_message(self, message):
        if is_dataclass(message):
            assert not isinstance(message, type)
            json_str = json.dumps(asdict(message), separators=(",", ":"))
            cls_name = getattr(type(message), "__message_type__", type(message).__name__)
            cls_id = self._message_types[cls_name].to_bytes(2, "big", signed = False)
            payload = _DATACLASS_MAGIC + cls_id + json_str.encode("utf-8")
        elif isinstance(message, bytes):
            payload = _GZIP_MAGIC + message
        else:
            payload = message.encode("utf-8")
        compressed = zlib.compress(payload, level=6, wbits=-15)
        return base64.b85encode(compressed).decode("utf-8")

    def write_message(self, key, message):
        """
        Write a message to the topic, assigning it to a partition based on the key.

        Messages may be:
        - `str`: stored as UTF-8 text,
        - `bytes`: stored as raw binary,
        - dataclass instances: encoded as JSON. Dataclasses must be registered with
        the Twain instance via `register_msg_cls()` before use.

        The key determines the target partition and the type is set on topic creating.

        The message is appended atomically to the selected partition file.

        Parameters
        ----------
        key : any
            Value used to select the partition. Must be compatible with the
            topic.
        message : str, bytes, or dataclass
            The message payload to write.

        Raises
        ------
        MessageTooLongError
            If the encoded message exceeds `MAX_MESSAGE_SIZE`.
        """
        encoded_key = key_to_base85(key, self.key_width)
        partition = self._partitioner(key, self._n_partitions)
        timestamp = encode_datetime(datetime.now(timezone.utc))
        msg_blob = self.encode_message(message)
        binary_msg = f"{encoded_key}{timestamp}{msg_blob}\n".encode("utf-8")
        if len(binary_msg) > MAX_MESSAGE_SIZE:
            raise MessageTooLongError("Message exceeds max message size: {len(binary_msg)} bytes > {MAX_MESSAGE_SIZE} bytes")
        atomic_append(self._active_file(partition), binary_msg)

    def _get_active_message_files(self):
        """Searches for the current active message files across all partitions"""
        topic_dir = self._topic_dir
        active_files = dict()
        message_files = list(topic_dir.iterdir())

        for partition in range(self._n_partitions):  
            chunk_str = self.chunk_str_now
            chunk_part_str = f"{partition}-{chunk_str}"
            active_file = [x for x in message_files if x.stem.split("_")[0] == chunk_part_str]
            if len(active_file) == 0:
                partition_files = [x for x in message_files if x.stem.split("-")[0] == str(partition)]
                if len(partition_files) == 0:
                    file_offset = 0
                else:
                    prev_file = max(partition_files)
                    with prev_file.open("r") as f:
                        lines = f.readlines()
                        prev_file_len = len(lines)
                    prev_file_offset = int(prev_file.stem.split("_")[1])
                    file_offset = prev_file_offset + prev_file_len
                new_active_file = topic_dir / f"{chunk_part_str}_{file_offset}.tmf"
                self._init_new_message_file(new_active_file)
                active_files[partition] = new_active_file
            elif len(active_file) == 1:
                active_files[partition] = active_file[0]
            else:
                raise TopicCorruptError(f"Multiple message files for the same chunk partition {partition}-{chunk_str}")
        return active_files, chunk_str

    def _init_new_message_file(self, new_active_file):
        new_active_file.touch()

    def _active_file(self, partition):
        """Returns the current file to be written to for a partition"""
        if self.chunk_str_now != self._chunk_str:
            self._active_files, self._chunk_str = self._get_active_message_files()
        return self._active_files[partition]

    def close(self):
        pass

__init__(twain, topic, partitioner=None)

Create a producer for a given topic.

Parameters

twain : Twain The TwainMQ instance backing this producer. topic : str The topic to write messages to. partitioner : callable, optional A function mapping (key, n_partitions) to a partition index. If omitted, a stable 64-bit hash partitioner is used.

Source code in src/twainmq/producer.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
def __init__(self, twain: "Twain", topic: str, partitioner = None): 
    """
    Create a producer for a given topic.

    Parameters
    ----------
    twain : Twain
        The TwainMQ instance backing this producer.
    topic : str
        The topic to write messages to.
    partitioner : callable, optional
        A function mapping `(key, n_partitions)` to a partition index.
        If omitted, a stable 64-bit hash partitioner is used.
    """
    super().__init__(twain, topic)
    if partitioner is None:
        partitioner = partition_hash64
    self._partitioner = partitioner
    self._active_files, self._chunk_str = self._get_active_message_files()

write_message(key, message)

Write a message to the topic, assigning it to a partition based on the key.

Messages may be: - str: stored as UTF-8 text, - bytes: stored as raw binary, - dataclass instances: encoded as JSON. Dataclasses must be registered with the Twain instance via register_msg_cls() before use.

The key determines the target partition and the type is set on topic creating.

The message is appended atomically to the selected partition file.

Parameters

key : any Value used to select the partition. Must be compatible with the topic. message : str, bytes, or dataclass The message payload to write.

Raises

MessageTooLongError If the encoded message exceeds MAX_MESSAGE_SIZE.

Source code in src/twainmq/producer.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
def write_message(self, key, message):
    """
    Write a message to the topic, assigning it to a partition based on the key.

    Messages may be:
    - `str`: stored as UTF-8 text,
    - `bytes`: stored as raw binary,
    - dataclass instances: encoded as JSON. Dataclasses must be registered with
    the Twain instance via `register_msg_cls()` before use.

    The key determines the target partition and the type is set on topic creating.

    The message is appended atomically to the selected partition file.

    Parameters
    ----------
    key : any
        Value used to select the partition. Must be compatible with the
        topic.
    message : str, bytes, or dataclass
        The message payload to write.

    Raises
    ------
    MessageTooLongError
        If the encoded message exceeds `MAX_MESSAGE_SIZE`.
    """
    encoded_key = key_to_base85(key, self.key_width)
    partition = self._partitioner(key, self._n_partitions)
    timestamp = encode_datetime(datetime.now(timezone.utc))
    msg_blob = self.encode_message(message)
    binary_msg = f"{encoded_key}{timestamp}{msg_blob}\n".encode("utf-8")
    if len(binary_msg) > MAX_MESSAGE_SIZE:
        raise MessageTooLongError("Message exceeds max message size: {len(binary_msg)} bytes > {MAX_MESSAGE_SIZE} bytes")
    atomic_append(self._active_file(partition), binary_msg)

TwainMQConsumer

Bases: TwainMQBase

A TwainMQ consumer.

Consumers may operate independently or as part of a consumer group. A standalone consumer can read messages but cannot commit progress. A consumer that belongs to a group (even a group of one) can commit offsets, allowing any consumer joining that group to resume from the last committed position.

The start_from parameter controls the initial read position only when: - the consumer is not part of a group, or - the group has no existing commit record.

Source code in src/twainmq/consumer.py 
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
class TwainMQConsumer(TwainMQBase):
    """
    A TwainMQ consumer.

    Consumers may operate independently or as part of a consumer group. A
    standalone consumer can read messages but cannot commit progress. A consumer
    that belongs to a group (even a group of one) can commit offsets, allowing
    any consumer joining that group to resume from the last committed position.

    The `start_from` parameter controls the initial read position only when:
    - the consumer is not part of a group, or
    - the group has no existing commit record.
    """
    def __init__(self, twain: "Twain", topic: str, start_from = "start", group = None):
        """
        Create a consumer for a given topic.

        Parameters
        ----------
        twain : Twain
            The TwainMQ instance backing this consumer.
        topic : str
            The topic to consume from.
        start_from : {"start", "now"}, optional
            Initial read position when no committed offset exists. `"start"`
            begins at offset 0; `"now"` begins at the current end of each
            partition.
        group : str or None, optional
            The consumer-group identifier. If `None`, the consumer operates
            independently and cannot commit offsets. If provided, the consumer
            joins the group and participates in consensus-based partition
            assignment and offset commits.
        """
        super().__init__(twain, topic)
        self._twain = twain
        self._topic = topic
        self._group = group
        self._rebal_ending = None
        self._consumer_id = None
        self._rebal_in_progress = None
        self._commit_record = dict()
        if group is None:
            self._partitions = [i for i in range(self._n_partitions)]
            if start_from == "start":
                self._consumerlets = {p: TwainMQConsumerlet(twain, topic, p, offset=0) for p in self._partitions}
            elif start_from == "now":
                self._consumerlets = {p: TwainMQConsumerlet(twain, topic, p, offset=None) for p in self._partitions}
            else:
                raise ValueError(f'start_from should be either "start" of "now", received: "{start_from}"')
        else:
            self._partitions = []      
            self._consumerlets = dict()      
            self._join_group(group, start_from)
        self._last_polled = 0
        self.heartbeat()

    def _create_consumer_group(self, group_topic, start_from):
        self._twain.create_topic(group_topic, key_type = "u32", partitions = 1, message_types = CONSUMER_GROUP_MESSAGE_SET)
        with self._twain.producer(group_topic) as gprod:
            if start_from == "start":
                init_commits = [(i, 0) for i in range(self._n_partitions)]
            elif start_from == "now":
                init_commits = [(i, -1) for i in range(self._n_partitions)]
            else:
                raise ValueError(f'start_from should be either "start" of "now", received: "{start_from}"')
            gprod.write_message(0, Commit(init_commits))

    def _join_group(self, group, start_from, retry=3):
        """Called when a consumer joins a group to handle id registration with the group"""
        group_topic = _group_topic_name(group)
        if not self._twain.topic_exists(group_topic):
            self._create_consumer_group(group_topic, start_from)
        gprod = self._twain.producer(group_topic)
        gcon = self._twain.consumer(group_topic, start_from = "start", group = None)
        self._group_producer = gprod
        self._group_consumer = gcon

        keys_taken = set()
        while (msg := gcon.poll()) is not None:
            keys_taken.add(msg.key)
            self._process_group_msg(msg)

        consumer_id = max(keys_taken) + 1
        join_key = random.getrandbits(64)
        gprod.write_message(consumer_id, Joined(join_key))

        while True:
            msg = gcon.poll()
            if msg is None:
                time.sleep(0.5)
            elif isinstance(msg.message, Joined):
                key = msg.message.tag
                if join_key == key:
                    self._consumer_id = consumer_id
                    self._begin_rebal()
                    return
                else:
                    gprod.write_message(consumer_id, AbortJoin(key))
                    if retry > 0:
                        time.sleep(random.random() * 0.1)
                        return self._join_group(group, start_from, retry = retry - 1)
                    else:
                        return
            else:
                self._process_group_msg(msg)

    def trigger_rebal(self):
        """
        Initiate a consumer-group rebalance.

        In TwainMQ, consumer-group membership is maintained through periodic
        consensus. Triggering a rebalance is lightweight and safe, and is typically
        done periodically to confirm that all consumers in the group are still
        active.

        If all consumers respond within the rebalance window, the rebalance becomes
        a no-op and existing partition assignments remain unchanged.
        """
        self._begin_rebal()

    def _begin_rebal(self):
        gprod = self._group_producer
        rebal_end_ts = (datetime.now(timezone.utc) + REBAL_LENGTH).timestamp()
        gprod.write_message(self._consumer_id, BeginRebal(rebal_end_ts))
        self.heartbeat()

    def _rebal_participate(self):
        gprod = self._group_producer
        gprod.write_message(self._consumer_id, RebalOffer(self._partitions, [i for i in range(self._n_partitions)]))

    def _end_rebal(self):
        gprod = self._group_producer
        gprod.write_message(self._consumer_id, EndRebal())
        self._confirm_rebal()

    def _confirm_rebal(self):
        gprod = self._group_producer
        assert self._rebal_in_progress is not None
        assignements = self._rebal_in_progress.get_assignments()
        self._partitions = assignements.get(self._consumer_id, [])
        self._consumerlets = {p: self._consumerlets.get(p, TwainMQConsumerlet(self._twain, self._topic, p, offset=self._commit_record.get(p,0))) for p in self._partitions}
        gprod.write_message(self._consumer_id, RebalConfirm(sorted(list(self._partitions))))
        self._rebal_in_progress = None

    def commit(self, partitions = None):
        if self._group is not None:
            if partitions is None:
                partitions = self._partitions
            commit_list = [(p, self._consumerlets[p]._offset) for p in partitions if p in self._consumerlets]
            print(commit_list)
            gprod = self._group_producer
            gprod.write_message(self._consumer_id, Commit(commit_list))

    def _handle_commit_msg(self, commit_msg):
        for p,o in commit_msg.partition_offsets:
            self._commit_record[p] = o

    def _process_group_msg(self, msg):
        if isinstance(msg.message, BeginRebal):
            if self._consumer_id:
                self._rebal_in_progress = RebalInProgress(msg.message.timestamp)
                self._rebal_participate()
        elif isinstance(msg.message, RebalOffer):
            if self._rebal_in_progress:
                self._rebal_in_progress.add_offer(msg.key, msg.message)
        elif isinstance(msg.message, EndRebal):
            if self._rebal_in_progress:
                self._confirm_rebal()
        elif isinstance(msg.message, Commit):
            self._handle_commit_msg(msg.message)
        else:
            pass

    def heartbeat(self):
        """
        Maintain membership in a consumer group and process group-management events.

        This method must be called periodically when using consumer groups; otherwise
        the consumer may be considered inactive and be removed from the group.

        Behaviour
        ---------
        - Processes any pending group-management messages (rebalance events, etc.).
        - Automatically invoked at the start of `poll()`.
        """
        if self._group is not None:
            while msg := self._group_consumer.poll():
                self._process_group_msg(msg)
            if self._rebal_in_progress and self._rebal_in_progress.rebal_ending < datetime.now(timezone.utc).timestamp():
                self._end_rebal()

    def poll(self):
        """
        Poll partitions in round-robin order and return the next available message.

        The consumer maintains a list of assigned partitions. Each call polls them
        starting from the partition after the one last successfully polled. The first
        non-`None` message returned by a partition is yielded.

        This call does not block.

        Returns
        -------
        message or None
            The next available message, or `None` if all partitions return no data.
        """
        self.heartbeat()
        n = len(self._partitions)
        for p_offset in range(n):
            partition_to_poll = self._partitions[(self._last_polled + p_offset + 1) % n]
            this_consumerlet = self._consumerlets[partition_to_poll]
            msg = this_consumerlet.poll()
            if msg is not None:
                self._last_polled = partition_to_poll
                return msg

    def poll_many(self, n=10):
        """
        Poll up to `n` messages and return them as a list.

        This is a convenience wrapper around `poll()`. Polling stops early if
        `poll()` returns `None`.

        Parameters
        ----------
        n : int, optional
            Maximum number of messages to retrieve.

        Returns
        -------
        list
            A list of messages, possibly shorter than `n`.
        """
        msgs = []
        for i in range(n):
            msg = self.poll()
            if msg is None:
                break
            msgs.append(msg)
        return msgs

    def __str__(self):
        return f"TwainMQConsumer(topic={self._topic}, group={self._group})"

__init__(twain, topic, start_from='start', group=None)

Create a consumer for a given topic.

Parameters

twain : Twain The TwainMQ instance backing this consumer. topic : str The topic to consume from. start_from : {"start", "now"}, optional Initial read position when no committed offset exists. "start" begins at offset 0; "now" begins at the current end of each partition. group : str or None, optional The consumer-group identifier. If None, the consumer operates independently and cannot commit offsets. If provided, the consumer joins the group and participates in consensus-based partition assignment and offset commits.

Source code in src/twainmq/consumer.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
def __init__(self, twain: "Twain", topic: str, start_from = "start", group = None):
    """
    Create a consumer for a given topic.

    Parameters
    ----------
    twain : Twain
        The TwainMQ instance backing this consumer.
    topic : str
        The topic to consume from.
    start_from : {"start", "now"}, optional
        Initial read position when no committed offset exists. `"start"`
        begins at offset 0; `"now"` begins at the current end of each
        partition.
    group : str or None, optional
        The consumer-group identifier. If `None`, the consumer operates
        independently and cannot commit offsets. If provided, the consumer
        joins the group and participates in consensus-based partition
        assignment and offset commits.
    """
    super().__init__(twain, topic)
    self._twain = twain
    self._topic = topic
    self._group = group
    self._rebal_ending = None
    self._consumer_id = None
    self._rebal_in_progress = None
    self._commit_record = dict()
    if group is None:
        self._partitions = [i for i in range(self._n_partitions)]
        if start_from == "start":
            self._consumerlets = {p: TwainMQConsumerlet(twain, topic, p, offset=0) for p in self._partitions}
        elif start_from == "now":
            self._consumerlets = {p: TwainMQConsumerlet(twain, topic, p, offset=None) for p in self._partitions}
        else:
            raise ValueError(f'start_from should be either "start" of "now", received: "{start_from}"')
    else:
        self._partitions = []      
        self._consumerlets = dict()      
        self._join_group(group, start_from)
    self._last_polled = 0
    self.heartbeat()

heartbeat()

Maintain membership in a consumer group and process group-management events.

This method must be called periodically when using consumer groups; otherwise the consumer may be considered inactive and be removed from the group.

Behaviour
  • Processes any pending group-management messages (rebalance events, etc.).
  • Automatically invoked at the start of poll().
Source code in src/twainmq/consumer.py 
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
def heartbeat(self):
    """
    Maintain membership in a consumer group and process group-management events.

    This method must be called periodically when using consumer groups; otherwise
    the consumer may be considered inactive and be removed from the group.

    Behaviour
    ---------
    - Processes any pending group-management messages (rebalance events, etc.).
    - Automatically invoked at the start of `poll()`.
    """
    if self._group is not None:
        while msg := self._group_consumer.poll():
            self._process_group_msg(msg)
        if self._rebal_in_progress and self._rebal_in_progress.rebal_ending < datetime.now(timezone.utc).timestamp():
            self._end_rebal()

poll()

Poll partitions in round-robin order and return the next available message.

The consumer maintains a list of assigned partitions. Each call polls them starting from the partition after the one last successfully polled. The first non-None message returned by a partition is yielded.

This call does not block.

Returns

message or None The next available message, or None if all partitions return no data.

Source code in src/twainmq/consumer.py 
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
def poll(self):
    """
    Poll partitions in round-robin order and return the next available message.

    The consumer maintains a list of assigned partitions. Each call polls them
    starting from the partition after the one last successfully polled. The first
    non-`None` message returned by a partition is yielded.

    This call does not block.

    Returns
    -------
    message or None
        The next available message, or `None` if all partitions return no data.
    """
    self.heartbeat()
    n = len(self._partitions)
    for p_offset in range(n):
        partition_to_poll = self._partitions[(self._last_polled + p_offset + 1) % n]
        this_consumerlet = self._consumerlets[partition_to_poll]
        msg = this_consumerlet.poll()
        if msg is not None:
            self._last_polled = partition_to_poll
            return msg

poll_many(n=10)

Poll up to n messages and return them as a list.

This is a convenience wrapper around poll(). Polling stops early if poll() returns None.

Parameters

n : int, optional Maximum number of messages to retrieve.

Returns

list A list of messages, possibly shorter than n.

Source code in src/twainmq/consumer.py 
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
def poll_many(self, n=10):
    """
    Poll up to `n` messages and return them as a list.

    This is a convenience wrapper around `poll()`. Polling stops early if
    `poll()` returns `None`.

    Parameters
    ----------
    n : int, optional
        Maximum number of messages to retrieve.

    Returns
    -------
    list
        A list of messages, possibly shorter than `n`.
    """
    msgs = []
    for i in range(n):
        msg = self.poll()
        if msg is None:
            break
        msgs.append(msg)
    return msgs

trigger_rebal()

Initiate a consumer-group rebalance.

In TwainMQ, consumer-group membership is maintained through periodic consensus. Triggering a rebalance is lightweight and safe, and is typically done periodically to confirm that all consumers in the group are still active.

If all consumers respond within the rebalance window, the rebalance becomes a no-op and existing partition assignments remain unchanged.

Source code in src/twainmq/consumer.py 
125
126
127
128
129
130
131
132
133
134
135
136
137
def trigger_rebal(self):
    """
    Initiate a consumer-group rebalance.

    In TwainMQ, consumer-group membership is maintained through periodic
    consensus. Triggering a rebalance is lightweight and safe, and is typically
    done periodically to confirm that all consumers in the group are still
    active.

    If all consumers respond within the rebalance window, the rebalance becomes
    a no-op and existing partition assignments remain unchanged.
    """
    self._begin_rebal()

TwainMQConsumerlet

Bases: TwainMQBase

A consumerlet is a simple single partition consumer. Usually you would not use a Consumerlet directly, rather use the Consumer container

Source code in src/twainmq/consumer.py 
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
class TwainMQConsumerlet(TwainMQBase):
    """A consumerlet is a simple single partition consumer.  Usually you would not use a Consumerlet directly, rather use the Consumer container"""
    def __init__(self, twain, topic, partition, offset=None):
        super().__init__(twain, topic)
        self._partition = partition
        self._seek_active_file(offset)

    def decode_message(self, message):
        compressed = base64.b85decode(message.encode("utf-8"))
        decoded = zlib.decompress(compressed, wbits=-15)
        if decoded.startswith(_GZIP_MAGIC):
            return decoded[1:]
        elif decoded.startswith(_DATACLASS_MAGIC):
            class_id = int.from_bytes(decoded[1:3], "big", signed=False)
            message_type = self._twain._msg_cls_registry[self._message_types_rev[class_id]]
            data = json.loads(decoded[3:])
            return dataclass_from_dict(message_type, data)
        else:
            return decoded.decode("utf-8")

    def _seek_active_file(self, offset = None):
        """Sets the file handle to the active file to read from and seeks to the end.

        If offset is None this is read from latest

        If the latest file is missing then it will return None.
        """
        part_files = self._list_partition_files()
        if offset is None or len(part_files) == 0:
            chunk_str = self.chunk_str_now
            head_file = [f for f in part_files if f.stem.split("-")[1].split("_")[0] == chunk_str]
            if len(head_file) == 0:
                self._current_file_handle = None
            elif len(head_file) == 1:
                offset = int(head_file[0].stem.split("_")[1])
                self._current_file_handle = head_file[0].open("r", encoding = "utf-8")
                while self._current_file_handle.readline():
                    offset += 1
                self._offset = offset
                self._chunk_str = chunk_str
            else:
                raise TopicCorruptError(f"Multiple message files for the same chunk partition {self._partition}-{chunk_str}")
        elif offset >= 0:
            offsets_files_chunks = sorted([(int(f.stem.split("_")[1]), f, f.stem.split("_")[0].split("-")[1]) for f in part_files])

            offsets = [o for o, _, _ in offsets_files_chunks]
            i = bisect.bisect_right(offsets, offset) - 1
            this_offset, current_file, chunk_str = offsets_files_chunks[max(i, 0)]

            self._current_file_handle = current_file.open("r", encoding = "utf-8")
            self._chunk_str = chunk_str
            self._offset = this_offset
            while self._offset < offset:
                line = self._current_file_handle.readline()
                if line != "":
                    self._offset += 1
                else:
                    break
        else:
            raise NotImplementedError("Seek back from end not yet implemented")

    def _list_partition_files(self):
        return [f for f in self._topic_dir.iterdir() if f.stem.split("-")[0] == str(self._partition)]

    def poll(self):
        """Polls for the next message without blocking.  Returns None if no new message available."""
        if self._current_file_handle is None:
            self._seek_active_file()
            if self._current_file_handle is None:
                return None
        msg_line = self._current_file_handle.readline()[:-1]
        if msg_line == "":
            if self.chunk_str_now == self._chunk_str:
                return None
            else:
                self._seek_active_file(self._offset)
                msg_line = self._current_file_handle.readline()[:-1]
                if msg_line == "":
                    return None
        key = base85_to_key(msg_line[:self._key_chars], self._key_width)
        timestamp = decode_datetime(msg_line[self._key_chars:self._key_chars+10])
        message = self.decode_message(msg_line[self._key_chars+10:])
        msg_tuple = Message(
            offset=self._offset,
            key=key,
            timestamp=timestamp,
            message=message
        )
        self._offset += 1
        return msg_tuple

    @property
    def partition(self):
        return self._partition

poll()

Polls for the next message without blocking. Returns None if no new message available.

Source code in src/twainmq/consumer.py 
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
def poll(self):
    """Polls for the next message without blocking.  Returns None if no new message available."""
    if self._current_file_handle is None:
        self._seek_active_file()
        if self._current_file_handle is None:
            return None
    msg_line = self._current_file_handle.readline()[:-1]
    if msg_line == "":
        if self.chunk_str_now == self._chunk_str:
            return None
        else:
            self._seek_active_file(self._offset)
            msg_line = self._current_file_handle.readline()[:-1]
            if msg_line == "":
                return None
    key = base85_to_key(msg_line[:self._key_chars], self._key_width)
    timestamp = decode_datetime(msg_line[self._key_chars:self._key_chars+10])
    message = self.decode_message(msg_line[self._key_chars+10:])
    msg_tuple = Message(
        offset=self._offset,
        key=key,
        timestamp=timestamp,
        message=message
    )
    self._offset += 1
    return msg_tuple