DefStream

CI and PyPI releases

The CI workflow runs on pull requests targeting master and pushes to master. Python 3.9 through 3.14 build the distributions, install the wheels in a clean virtual environment, and run this repository's tests against RabbitMQ 4.1 and Redis 7.4 service containers. The core JavaScript UI tests also run. Only a successful master push can publish to PyPI.

The workflow checks out the current repository's event commit and the other two yong5eon/DefStream* repositories at master, recording all commit IDs in the log. Those repositories must be accessible and have a master branch. All three packages are built locally, so the first release does not depend on existing PyPI releases. This tests the source combination, not compatibility with previously published dependency versions.

For a push, VERSION must contain major.minor.patch; the build replaces its patch component with github.run_number (for example, 0.1.0 becomes 0.1.42 on run 42). PR builds retain the checked-in version. No version commit, PR, or Git tag is created. Change the major/minor in VERSION when starting a new release series. Run numbers can have gaps; renaming/replacing this workflow or switching an existing package to this scheme requires checking for collisions.

After every Python matrix entry passes, the publish job downloads the exact Python 3.13 distributions already built and tested. A fully published version is skipped only on a rerun; first-attempt collisions and partially published versions fail explicitly. Network errors are not treated as missing releases. Concurrent runs are serialized without cancelling the active run; GitHub may supersede an older pending run when more pushes arrive.

Before enabling automatic releases:

1. In each repository, add PYPI_API_TOKEN under Settings > Secrets and variables > Actions, or as a secret in the pypi environment. Use the appropriate PyPI project scope once the project exists. Never commit tokens.
2. Configure the pypi environment to allow deployments only from master. Do not require manual approval if fully automatic deployment is desired.
3. Protect master, require PRs, and require all six Test Python 3.x checks. Do not require Publish PyPI, which intentionally does not run on PRs.
4. Keep the workflow filename/name stable and review changes to workflows and packaging carefully: merging to master authorizes a public release.

Dependencies still need to be published before end users can install adapters from PyPI. For the initial release, publish DefStream before its adapters.

Just def.
Let workers do the work.
You don't need to await this.

Installation

pip install DefStream
pip install "DefStream[RabbitMQ]"
pip install "DefStream[Redis]"
pip install "DefStream[RabbitMQ,Redis]"

The base package has no broker client dependencies and does not include adapter source files. Extras install the separate DefStream-RabbitMQ and/or DefStream-Redis distributions, which depend on pika and redis, respectively. You can also install the adapters directly:

pip install DefStream-RabbitMQ
pip install DefStream-Redis

The Python import paths remain DefStream.Broker.Implements.RabbitMQ and DefStream.Broker.Implements.Redis. For local development of all three projects, see the editable installation instructions below.

The project is now named DefStream. Distribution/import names use DefStream, Prometheus metric names use defstream_, and integration-test environment variables use DEFSTREAM_. Redis's default consumer group is now DefStream; when continuing an existing deployment, explicitly keep its previous group name to preserve consumer-group position and pending messages. Queue names explicitly configured by applications are unchanged.

from DefStream.Worker import Worker
from DefStream.Broker.Implements.RabbitMQ import Connection as RabbitConnection

There is no combined DefStream.Broker.Broker interface. Implementations expose a concrete Connection class without a Broker compatibility alias.

Create broker connections directly with the implementation's configuration and call connect() before use. Call close() when finished. Connection registries or helpers, if needed, belong in application code.

RabbitMQ resource methods use snake_case: create_exchange, delete_exchange, bind_exchange_to_exchange, unbind_exchange_to_exchange, create_queue, delete_queue, bind_exchange_to_queue, unbind_exchange_to_queue, create_stream, and delete_stream. The former camelCase names are removed.

Connection capabilities

DefStream.Broker.Connection defines independent typing.Protocol contracts, also exported from DefStream.Broker. Implementations explicitly inherit only the capabilities they provide:

Protocol	Contract	RabbitMQ	Redis
Connection	connect(), close()	Yes	Yes
GetExchange	exchange(name) -> Sendable	Yes	No
GetQueue	queue(name) -> Sendable	Yes	Yes
GetStream	stream(name) -> Sendable	Yes	Yes
GetConsumer	consumer(handler=None) -> Consumer	Yes	Yes

For example, RabbitMQ's Connection inherits Connection, GetExchange, GetQueue, GetStream, and GetConsumer. Redis omits GetExchange and uses its Queue for both GetQueue and GetStream. Unsupported methods are absent rather than stubs raising NotImplementedError.

These are plain Protocols without generic type parameters. Resource factories promise the shared Sendable contract; concrete adapter methods retain their specific return annotations for adapter-specific operations. The common resource-name argument is positional-only to accommodate existing adapter argument names; call provider.queue('name') through a generic contract. Concrete implementations retain their existing keyword arguments and can add optional settings. The contracts intentionally do not promise arbitrary **kwargs. Missing connection/factory implementations prevent instantiation of explicit subclasses through @abstractmethod.

DefStream.Worker.WorkerConnection composes only Connection and GetConsumer. Worker does not require queues, streams or exchanges. Application connections can inherit these protocols explicitly or satisfy them structurally without inheritance. Protocols are for static checking, not isinstance capability checks.

Events (including the RPC failed-request wrapper), error policies and Redis JSON codecs also explicitly inherit their existing Event, EventError, Encoder and Decoder protocols. Message objects explicitly inherit Sendable, Callable, Poppable and/or Readable as appropriate. Consumer, Run, configuration and lifecycle-hook ABCs are retained: this change does not replace implementation base classes or introduce new contracts for every callable.

Message capabilities

Common message I/O contracts now live in DefStream.Broker.Protocol and are also exported from DefStream.Broker:

Protocol	Operation	Timeout
Sendable	send(body)	Implementation-specific
Callable	call(body, timeout=5)	Seconds
Poppable	pop(timeout=None)	Milliseconds, preserving the existing API
Readable	read(offset=0, timeout=None)	Milliseconds, preserving the existing API

These are structural typing.Protocol contracts, not runtime capability checks. Implementations can expose additional optional broker-specific arguments. The former common Queue, Stream, and Exchange base classes and their modules have been removed. Concrete RabbitMQ classes retain their names: Exchange is Sendable/Callable, Queue is Sendable/Callable/Poppable, and Stream is Sendable/Readable. If typing.Callable is needed alongside the message Callable, import the typing symbol as PyCallable. Event, EventHandler, and EventError live together in Broker.Event and are exported from DefStream.Broker. EventHandler is a Protocol with __call__(event: Event) -> None; explicitly inheriting handlers must implement it. The former standalone Broker.EventHandler module has been removed. EventError is a failure-policy contract, not an I/O capability.

RabbitMQ RPC

Both RabbitMQ Queue and Exchange explicitly inherit Callable and implement call() directly, including request publication, reply matching, and timeout handling; there is no shared RabbitMQ Rpc module. Obtain them through broker.queue() or broker.exchange() so the decoder and dedicated RPC connection parameters are available.

For Exchange RPC, event selects the routing key, just as with send():

result = broker.exchange('operations').call(
	{
		'a': 10,
		'b': 20,
	},
	timeout=5,
	event='add',
)

Bind the exchange to the server queue before calling. An omitted event uses the empty routing key. If the exchange routes to multiple queues, each may execute the request; the client returns the first matching response, not an aggregate. Use a single destination for ordinary RPC to avoid duplicate side effects. Unroutable requests raise UnroutableError.

Use a queue obtained from broker.queue(name) for synchronous request/reply:

result = broker.queue('add').call(
	{
		'a': 10,
		'b': 20,
	},
	timeout=5,
)

An ordinary Run serves both one-way messages and RPC requests. For RPC, its return value (including None) is published after RunComplete hooks succeed, then the request is ACKed. Run/hooks still execute in the selected pool; response publication and message settlement execute on the connection owner thread. No connection or event object is passed to ProcessPool children.

Each call opens its own connection and an exclusive, auto-delete response queue, subscribes before publishing, and matches responses by correlation ID. It closes that connection and response queue on success, timeout, or failure. This initial implementation favors simple ownership over connection reuse: there is no persistent Rpc client or asynchronous/multiplexed call API yet. The queue object's existing connection/channel must still only be used by its owner thread. Do not block the Worker consumer thread waiting for a call to itself; invoke RPC from a separate client.

Requests use reply_to, correlation_id, and a TTL based on the remaining timeout. Calls raise TimeoutError if no response arrives before the deadline; this does not cancel work already delivered. Connection establishment uses bounded socket/stack timeouts. Blocking broker operations and connection cleanup mean the timeout is not a hard wall-clock limit on the entire call. Both request and response publishers use confirms and mandatory routing. An unavailable request queue raises Pika's UnroutableError instead of waiting.

Requeue sends no final response, allowing retry. When the selected EventError calls reject(), the server sends an error envelope before rejecting the request (and preserving configured dead-letter routing). The client raises DefStream.Broker.RemoteError with code ProcessingError and a generic message; server exception details stay in server logs. Custom EventError implementations must settle via the supplied event's ack/nack/reject methods.

A response destination may disappear when a caller times out. An unroutable response is logged as a warning and the original request is settled normally; the Worker keeps running. Other response publication/serialization failures propagate without ACK/Reject, close the Worker, and leave unconfirmed requests for broker redelivery. Response publication and request ACK are not atomic: crashes can cause duplicate execution/replies, so side-effecting Runs must be idempotent. RPC does not provide exactly-once delivery.

The RabbitMQ worker sample shares its Add and Divide tasks in sample/Worker/RabbitMQ/Tasks.py. To try RPC, start sample/Worker/RabbitMQ/Worker.py, then run sample/Worker/RabbitMQ/Client.py in another terminal. The client uses request(body, task=..., broker=...) with these shared tasks and shows successful addition/division and a rejected division by zero. For a standalone broker-only example, run sample/Broker/Implements/RabbitMQ/Sample.RPC.py. It starts a responder on its own connection/thread, demonstrates Queue and Exchange RPC plus RemoteError, then removes only its uniquely named queue and exchange. A local RabbitMQ server is required, but no separate Worker process is needed.

Redis Streams

The Redis adapter uses Redis Streams, not Pub/Sub or lists. Install the Redis extra and use Redis 6.2 or newer (including XAUTOCLAIM support):

from DefStream.Broker.Implements.Redis import (
	Connection,
	Configuration,
)

broker = Connection(Configuration(
	host='127.0.0.1',
	port=6379,
))
broker.connect()
try:
	queue = broker.queue('add')
	queue.send({
		'a': 10,
		'b': 20,
	})
	event = queue.pop(timeout=1000)
	if event is not None:
		print(event.id, event.body)
		event.ack()
finally:
	broker.close()

broker.queue() and broker.stream() return the same Queue, implementing Sendable, Poppable and Readable. Resource methods are create_queue, delete_queue, create_stream, and delete_stream. Creating a queue creates the stream and consumer group; sending or subscribing can create them lazily. Group creation starts at 0-0, so messages published before subscription are not lost. Use the concrete Connection class directly. Exchanges and RPC are not supported; Redis Connection neither inherits GetExchange nor exposes exchange().

Configuration supports username/password (including password-only auth), database, TLS (secure=True), connection limit (max, default 10), and socket timeout (timeout, seconds, default 5). The default ConnectionType.Pool uses redis-py's managed connection pool; ConnectionType.Cluster uses RedisCluster and requires database 0. Cluster mode is implemented but the local integration suite exercises standalone Redis only. Connection failures propagate from connect(), which verifies connectivity with PING.

Bodies use a single stream field named body containing UTF-8 JSON, preserving nested mappings, lists, numbers, booleans and null. Pass callable encode and decode configuration values for a different byte codec. Legacy field-per-value streams are not wire-compatible and are not decoded using eval.

pop(timeout=...) waits for one group delivery; timeout is in milliseconds. None or zero waits indefinitely, matching the existing RabbitMQ convention. Positive timeouts return None when no delivery arrives. read(offset=0, timeout=...) yields history followed by new events until its total millisecond timeout; None or zero tails indefinitely. Redis offsets may be stream-ID strings, such as '1720000000000-0'; the offset is exclusive. History reads do not participate in consumer groups, so their events cannot be acknowledged/requeued. Closing the iterator stops reading.

Workers accept Subscription(queue='add', group='DefStream') from the Redis adapter; a queue-name string or equivalent mapping also works. Consumers within one group compete for messages; separate groups each receive the stream. broker.consumer(qos=1) limits total unacknowledged local deliveries across all subscriptions. Owner-thread callbacks handle pool completion and shutdown, including ThreadPool and ProcessPool workers. Redis Run return values are acknowledged normally but are not sent back as RPC responses.

• ack() and reject() use XACK without deleting history. Reject has no built-in dead-letter destination.
• Default failure handling uses Requeue. nack() atomically appends a copy to the stream and acknowledges the original pending entry using a single-key Lua script. The retry has a new stream ID and is visible to all groups, not only the failing group. Separate retry streams are needed for group-isolated retry.
• Consumers reclaim abandoned pending entries with XAUTOCLAIM after 60 seconds of idle time by default. Set reclaim_idle on broker.consumer() to a larger millisecond value for long tasks, or None to disable automatic recovery. Long-running active tasks can otherwise be reclaimed by another consumer. The default Worker uses the default consumer settings.
• Shutdown does not acknowledge unfinished work. Connection failures and decoding/handler/callback errors propagate; there is no automatic reconnect. Unconfirmed entries remain pending for later reclamation.

Delivery is at least once: make side effects idempotent. History is retained after settlement and retries add entries; configure Redis retention externally without trimming pending entries. The adapter does not silently trim/delete messages or destroy groups.

Run sample/Broker/Redis/Sample.py for a finite send/pop/history demo. sample/Broker/Redis/Sample.Stream.py uses broker.stream() to publish three events, read from the beginning, and resume after the first event's stream ID. These history reads do not require ACK. Each read uses a 1000 ms timeout, and the sample deletes only its uniquely named stream and closes the connection. sample/Worker/Redis/Sample.py uses Prepare to create each registered queue and consumer group and publish five messages per Run before consuming add and divide. Operands range from 0 to 99; the first Divide message always uses a zero divisor to demonstrate logged ZeroDivisionError and Reject without endless requeue. Other divisors are random and may also be zero. Successful messages continue processing. It runs until Ctrl+C, then drains work and runs Cleanup. Cleanup logs completion without deleting shared queues or stream history. You can publish more {'a': 10, 'b': 20} messages from another process. Enable the real Redis integration tests with DEFSTREAM_REDIS_TESTS=1; they use localhost:6379 and create/delete only uniquely named test streams.

Workers

Inspector: HTTP health checks and metrics

Inspector uses only the Python standard library: a threaded WSGI HTTP server managed by the Worker. It starts before broker connection/setup, and closes after pool draining, cleanup and broker closure, including startup failures. An occupied HTTP port fails Worker startup explicitly. Inspector is optional; workers without one keep their existing behavior.

from DefStream.Inspector import Inspector
from DefStream.Inspector.Extensions import (
	HealthCheck,
	Metrics,
)
from DefStream.Worker import (
	ThreadPool,
	Worker,
)

worker = Worker(
	connection,
	ThreadPool(10),
	inspector=Inspector(
		host='127.0.0.1',
		port=9999,
		exts=(
			HealthCheck('/status'),
			Metrics('/metrics'),
		),
	),
)
worker.add(Sample)
worker.run()

• GET /status returns JSON with status. It returns 200 only after subscriptions are ready and the consumer loop answers a thread-safe callback within HealthCheck(timeout=1) seconds (default 1). Idle workers are healthy. Starting, stopping, stopped and unresponsive workers return 503 while HTTP remains open. A terminated process/server fails the HTTP connection instead. The probe checks consumer-loop responsiveness, not every dependency or a hung individual task; choose a timeout above normal consumer-loop latency. Repeated/concurrent checks share an outstanding probe rather than filling a stalled consumer's callback queue.
• GET /metrics returns Prometheus text format, 200 even when not ready, so failures and shutdown can still be observed. Metrics include defstream_worker_running, defstream_worker_uptime_seconds, defstream_tasks_active, defstream_tasks_total (success/failure), and defstream_task_duration_seconds (summary count/sum). The running gauge describes lifecycle state; use /status for the responsiveness probe. Task labels use module-qualified class names; no message body/ID is exposed. Counters describe completed execution attempts, including retries and hooks, not confirmed ACKs or unique messages. Invalid bodies, pool-full requeues and failed submissions do not count as task executions. Duration includes pool wait and hooks. Metrics are local to one Worker process and reset on restart.
• Paths are explicit extensions; there are no default routes. Both endpoints support HEAD. Unknown paths return 404 and unsupported methods return 405. Extensions default to GET/HEAD; request-aware extensions can declare methods and override handle(inspector, environ) without changing existing endpoints. Additional endpoints can inherit Extension and return Response, both exported from DefStream.Inspector. Built-in extensions are exported from DefStream.Inspector.Extensions.
• Each extension owns its behavior and state: HealthCheck owns the consumer probe, lock and timeout; Metrics owns counters, execution timing and output. Inspector owns only HTTP serving, Worker attachment, routing and event dispatch. Unregistered extensions do not collect data or create probe state. Extension.start(inspector) initializes an extension before HTTP startup. begin(task) returns an extension-owned context; end(task, context, success) receives it on completion. success=None means submission was rejected and must not count as a completed attempt. These hooks execute in the parent process and may run on different threads; extensions must synchronize state.
• Inspector is attached to one Worker and cannot be restarted or shared. Task execution remains in the existing ThreadPool/ProcessPool; the HTTP server and metric aggregation stay in the parent process.
• This is a small operational HTTP interface, not a public web server. No TLS is provided. Health, metrics and Specifications routes are unauthenticated; the optional Console extension supports an optional Bearer token for task data and execution. Keep Inspector on loopback or a trusted management network.

The RabbitMQ and Redis Worker samples enable /status and /metrics on 127.0.0.1:9999; use different ports if running both at once. For Docker, use host='0.0.0.0' to expose it outside the container (restrict network access). A health command inside the container needs no curl dependency:

HEALTHCHECK --interval=10s --timeout=3s --start-period=15s --retries=3 CMD python -c "from urllib.request import urlopen; urlopen('http://127.0.0.1:9999/status', timeout=2).close()"

Live dashboard

Register Dashboard together with exactly one Metrics extension:

from DefStream.Inspector.Extensions import (
	Dashboard,
	Metrics,
)

inspector = Inspector(exts=(
	Metrics('/metrics'),
	Dashboard(
		'/dashboard',
		title='DefStream',
		description='Worker activity',
	),
))

Open /dashboard for a lightweight, dependency-free status page. Both Worker samples register this route. The page reads the existing Metrics collector; there is no second collector, broker connection or server-side history. title defaults to DefStream; optional description is the subtitle. HTML changes take effect on refresh without restarting.

• Shows worker lifecycle, uptime, in-flight submissions, cumulative success and failure counts, total throughput, and per-task throughput and average time.
• Select Last second or Last minute for real server-side 1/60-second completion windows, not a per-second value multiplied by 60. Windows end at the last fully elapsed monotonic second (less than one second behind live totals). The minute window advances every second, not once per calendar minute. Startup windows include only data since startup, without extrapolation. History is available immediately on page load and after reconnect.
• Selected-window metrics include completed success/failure counts, failure percentage, average and maximum duration, both overall and per task. An empty window shows zero completions and -- for failure percentage/average/maximum. Oldest in-flight age is measured from submission, including pool wait.
• Averages use the duration of attempts completed in the selected window, including pool wait and hooks; these are not p95 latency. Retries count as separate attempts; pool-full/rejected submissions do not count as completed. In flight includes submitted work waiting in the pool, not broker queue depth.
• Lifecycle is not consumer responsiveness. Use HealthCheck for health probes. Requests time out after five seconds; failed polls mark totals as stale and clear current rates. Pause/resume is available; hidden tabs stop polling.
• GET /dashboard with Accept: application/json returns a detached snapshot (instance, sampled_at, uptime, running, tasks). Tasks are keyed by module-qualified name with cumulative active, success, failure, duration (seconds), oldest_active (seconds), and windows keyed by "1"/"60" with success, failure, duration, max_duration. window_end is the monotonic exclusive end of both windows. Metrics keeps at most 61 one-second buckets per task (including the current incomplete second), not individual completion records. Pending submission timestamps are retained only until completion. Prometheus also exposes defstream_tasks_window by window_seconds/outcome, defstream_task_window_duration_seconds_sum, defstream_task_window_duration_seconds_max, and defstream_task_oldest_active_seconds. No p95 is collected. Prometheus and Dashboard use the same lock-protected snapshot.
• Like Metrics, Dashboard is unauthenticated. Restrict access at the application, network or proxy layer. It shows one Worker only, resets on restart, and is not a replacement for Prometheus/Grafana history or alerting.

Request/response specifications

DefStream.Descriptor and DefStream.Validator reuse the local Liquirizia Description/Validator APIs under the repository's GPLv3 license. Descriptor has no DataModel dependency: manual schemas and dataclass ToSchema/ToObject remain available. The original temporary sources are unchanged. Existing names and implementation style are retained in the ported modules to minimize drift. DataModel-only conversions and IsDataModel are not included.

from DefStream.Descriptor import (
	Integer,
	Object,
	Properties,
)
from DefStream.Validator import Validator
from DefStream.Validator.Patterns import (
	IsInteger,
	IsObject,
)
from DefStream.Validator.Patterns.Object import (
	IsMappingOf,
	IsRequiredIn,
)
from DefStream.Worker import (
	Run,
	Specification,
)

class Add(
	Run,
	parameters={'queue': 'add'},
	request=Specification(
		schema=Object(
			properties=Properties(
				a=Integer('First operand'),
				b=Integer('Second operand'),
			),
		),
		validator=Validator(IsObject(
			IsRequiredIn(
				'a',
				'b',
			),
			IsMappingOf({
				'a': IsInteger(),
				'b': IsInteger(),
			}),
		)),
		example={
			'a': 10,
			'b': 20,
		},
	),
	response=Specification(
		schema=Integer('Sum'),
		validator=Validator(IsInteger()),
		example=30,
	),
):
	def run(
		self,
		a: int,
		b: int,
	):
		return a + b

Specification accepts schema, validator, example, description, and format (default application/json). For example, format='text/plain' sets the message's AsyncAPI contentType. Request and response can declare different media types. This is contract metadata; it does not change broker codecs or the JSON content type of the specifications HTTP endpoint. Configure compatible broker encoders/decoders separately when using a different wire format. schema can be a Descriptor value, Schema/condition, JSON-schema mapping, or dataclass type. It describes the wire data; it does not automatically validate or instantiate a model. Use Validator for runtime rules and transformations. Custom validator rules are not reverse-engineered into JSON Schema. Examples are documentation values, not automatically executed or validated.

Execution order is setup hooks, request validation/transformation, run, response validation/transformation, completion hooks, then RPC response/ACK. Request validation must return a mapping with string keys; specified tasks accept keyword arguments. Output can be any existing broker-serializable value. The returned validator value is used, not discarded. Validation failures invoke RunError hooks and the existing on_error policy; configure Reject where permanent invalid messages must not be requeued indefinitely. Request and response contracts are inherited. Tasks without contracts retain their behavior. The send/request helpers do not run the validators again.

Necessary changes from the source: AllOf dumps as allOf; string min/max dump as minLength/maxLength; unsupported schema values/references fail explicitly. Conversion patterns use ast.literal_eval, not eval: numeric literal strings still work but executable expressions are rejected. Validate supports functions without defaults and positional-only arguments.

AsyncAPI JSON

from DefStream.Inspector.Extensions import Specifications

# Add to Inspector(exts=(...)).
specifications = Specifications(
	'/asyncapi.json',
	version='1.0.0',
	description='Worker task contracts',
	format='application/json',
)

Specifications.format sets the document's defaultContentType and defaults to application/json. Messages without a request Specification inherit this default; each declared Specification's format remains an explicit message-level override (it also defaults to JSON). This does not change broker codecs or the JSON HTTP response format of this endpoint.

GET /asyncapi.json serves an AsyncAPI 3.1.0 document generated from that Worker's registered tasks, without a global registry or broker connection. The document is available during startup/shutdown while HTTP remains open. It describes Worker-side receive operations, queue addresses, input message schemas/examples. Operation descriptions come from request.description; when it is None, the field is omitted without falling back to task docstrings. Unspecified inputs have an object payload with no additional constraints. Queue names can come from strings, subscription mappings or subscription objects; missing addresses fail explicitly. Operation keys are task class names (without modules). Receive channel keys are the actual queue names, preserved without replacement; JSON Pointer references escape / and ~. Request messages are defined directly under each channel, referenced as #/channels/<queue>/messages/request. Specification has no channel option; subscription parameters determine the queue. Output contracts generate a standard operation reply, referencing the same channel and its response message at #/channels/<queue>/messages/response. Request and response definitions live together under the channel's messages; the receive operation selects only request, and its reply selects only response. No separate .completion channel is generated. Duplicate operation or queue channel keys fail explicitly. This describes the declared request/reply contract, not broker capabilities: declaring an output does not turn Redis into RPC, and ordinary sends do not request replies. Dynamic RabbitMQ reply address extraction, correlation and transport bindings are not modeled yet. No server credentials are exposed. This extension serves JSON only; the optional Console extension provides the interactive task UI.

The RabbitMQ Worker sample declares contracts for Add and Divide and serves them at /asyncapi.json. The Redis sample exposes the same endpoint, with unconstrained inputs for tasks that have no contract yet.

Interactive task console

Console provides a self-contained Swagger-style screen: registered tasks, descriptions, request/response schemas, editable JSON examples and execution results. It has no CDN or JavaScript package dependencies. It is not Swagger UI (which consumes OpenAPI); it displays the same generated AsyncAPI contracts. Schemas are displayed as expandable field trees with types, required markers, descriptions and constraints. Nested objects, array items and composition branches are shown separately; the complete raw JSON schema remains available. Local schema references are expanded with recursion detection; external references are labeled without fetching them. Deep models are limited to 12 nesting levels in the tree, with the full schema preserved in the raw view. The bundled HTML is read on each page request with caching disabled. Editing Console.html takes effect on browser refresh without restarting the Worker; Python code and task contract changes still require a restart.

from DefStream.Inspector.Extensions import Console
from DefStream.Broker.Implements.RabbitMQ import (
	Connection,
	Configuration,
)

def create_connection() -> Connection:
	return Connection(Configuration(
		host='127.0.0.1',
		port=5672,
		username='guest',
		password='guest',
	))

# Add to Inspector(exts=(...)).
console = Console(
	'/console',
	broker=create_connection,
	title='DefStream',
	description='Task contracts & interactive messaging',
	timeout=5,
)

Console.title defaults to DefStream and sets both the HTML document title and header name without a suffix. Optional description is the subtitle; omitting it hides the subtitle. Both values are escaped as plain text. After loading tasks, the document version is displayed beneath the subtitle. Console uses the first registered Specifications extension, including its version, description and media type settings. Without one, it generates a default document with version 1.0.0. Inspector has no name option: health responses contain status only, metrics have no worker-name label, and Specifications uses DefStream as its info title.

The factory must return a new, independently owned producer connection on every call, never the Worker's consumer connection or a shared singleton. The extension connects for execution and closes in finally, including failed connect/publish/RPC. Catalog loading inspects a separate, unconnected producer's RPC capability and closes it. It never declares/deletes queues or calls Run directly. Validators run on the actual Worker, not again in the HTTP thread.

• Open /console. With token omitted or None, no authentication is required: tasks load automatically and the token controls are hidden. To enable Bearer authentication, pass token='your-random-token-at-least-16-characters', enter it in the page, then choose Load tasks. HTML is a public, static shell with no embedded token or task data; when authentication is enabled, task metadata and execution both require Authorization: Bearer <token>. Tokens must have at least 16 characters; use a randomly generated strong token. The page keeps the token only in memory, never browser storage; Clear token removes it.
• Send invokes the existing task send helper and displays the message ID. RPC invokes request and displays the result. RPC is enabled only when the producer supports Requestable (currently RabbitMQ, not Redis); it does not depend on a response contract. Redis still supports Send.
• GET /console with Accept: application/json returns the catalog (authenticated only when a token is configured). POST /console accepts JSON {"task":"Add","mode":"send","body":{"a":10,"b":20}}; use "mode":"request" for RPC. Only registered, unambiguous task names are accepted. Body must be a JSON object; arbitrary queues cannot be selected.
• Unknown/invalid inputs return 400, unauthorized calls 401, cross-origin requests 403, bodies above 64 KiB 413, non-JSON requests 415, excess concurrent calls 429 (four slots), stopped/not-ready Workers 503, unsupported RPC 501, remote task errors 502 and RPC timeouts 504. Infrastructure failures are logged and return a generic 500 without exposing connection details.
• Timeouts do not cancel already delivered work. Send/RPC are not retried by the UI. These operations can produce real side effects; use test queues and idempotent tasks. Configure broker connection/socket timeouts as well as RPC timeout; metadata Specification.format does not change producer codecs.
• No CORS is enabled. Calls with an Origin must match the HTTP Host. Remote access must use HTTPS through a trusted reverse proxy to protect the token; do not expose the standard-library WSGI server directly to the Internet. Other Inspector extensions remain unauthenticated.

Both Worker samples enable /console without a token. Authentication, access control, token configuration and secret management are the application's responsibility.

Task-directed messages and declarations

declare, send and request are exported from DefStream.Worker. They resolve the destination from a concrete Run class's parameters; the sending application does not need a Worker or global task registry. Bodies must be mappings with string keys, matching run(**body). A scalar Run result must be wrapped in a mapping before forwarding to another task.

Connections are borrowed: callers must connect and close them. Neither helper connects, declares destinations, nor closes the passed connection, including on failure. One connection can publish many records:

from DefStream.Worker import (
	declare,
	send,
	request,
)
from DefStream.Broker.Implements.RabbitMQ import (
	Connection,
	Configuration,
)

connection = Connection(Configuration(
	host='127.0.0.1',
	port=5672,
	username='guest',
	password='guest',
))
connection.connect()
try:
	# Sample is an application Run class with subscription parameters.
	declare(
		task=Sample,
		broker=connection,
	)
	for index in range(1000):
		send(
			{
				'a': index,
				'b': 1,
			},
			task=Sample,
			broker=connection,
		)
	response = request(
		{
			'a': 10,
			'b': 20,
		},
		task=Sample,
		broker=connection,
		timeout=5,
	)
finally:
	connection.close()

send returns the published message ID. RabbitMQ uses one reusable publisher channel per connection, publisher confirms and mandatory routing; an absent destination raises rather than silently dropping a task. Redis uses XADD. The former task-directed emit name has been removed without an alias. emit is reserved for a future exchange/event publishing API, not implemented yet. RabbitMQ request delegates to Queue RPC, returning its result or propagating TimeoutError/RemoteError/publication errors. Redis request raises NotImplementedError. RPC still opens a dedicated request/reply connection per call internally; the caller-owned connection remains open. This is not pooled or multiplexed RPC.

worker.tasks replaces worker.runs and contains registered Run classes. Declarations remain explicit, not a side effect of subscribing or sending:

class Declare(WorkerSetup):
	def __call__(
		self,
		worker: Worker,
	) -> None:
		for task in worker.tasks:
			declare(
				task=task,
				broker=worker.broker,
			)

Pass setups=Declare() to Worker. Its broker is already connected before setup runs. declare(task=task, broker=connection) delegates to connection.declare(task.parameters) without opening or closing the connection. It can be used without a Worker; there is no worker.declare() method. RabbitMQ declares durable queues and, when error is provided, the durable DLQ and dead-letter routing through the default exchange. Redis creates the stream and named consumer group. Existing incompatible RabbitMQ declarations fail explicitly; no queue is deleted/recreated automatically. This initial task declaration API provisions ordinary queues, not RabbitMQ stream queues or exchange bindings.

Adapters explicitly implement the optional Declarable, TaskSender and Requestable protocols in Broker.Connection (also exported from Broker). These three protocols support runtime capability checks so helpers can report unsupported operations. Other connection protocols remain static contracts. A normal Worker still requires only connect/close/consumer; declaration is optional.

Completion hooks can forward work:

class SendNext(RunComplete):
	def __call__(
		self,
		context,
		completion,
	) -> None:
		connection = Connection(Configuration(
			host='127.0.0.1',
			port=5672,
			username='guest',
			password='guest',
		))
		connection.connect()
		try:
			for record in completion:
				send(
					record,
					task=Sample2,
					broker=connection,
				)
		finally:
			connection.close()

Here completion is an iterable of mappings. For one mapping, call send once instead. Use completes=SendNext() on the source Run. Both destinations must already be declared. Hooks execute inside the pool thread/process: open their producer connection there, never share the consumer connection or a live connection across processes/threads. For ProcessPool, hook and Run classes must be defined at module scope. Connections can be reused within a batch; reuse across multiple Runs can be managed by application PoolSetup/PoolCleanup hooks. Keep RabbitMQ heartbeats serviced during long idle/compute periods.

Hook publication failures propagate as task failures and prevent success ACK. Earlier publications in a partially completed batch are not rolled back. Publication and source ACK are not atomic: retries/crashes may duplicate downstream messages, so recipients must be idempotent. Blocking requests from a hook to the same exhausted pool can deadlock until timeout; use separate capacity/workers for such dependencies.

Workers receive a broker and a bounded execution pool. Register Run classes explicitly; importing or defining a Run does not register it globally. EventContext, EventProperties, EventRunner, worker Factory, Invoker, and automatic module loading have been removed.

from DefStream.Worker import (
	declare,
	Run,
	ThreadPool,
	Worker,
	WorkerSetup,
)
from DefStream.Broker.Implements.RabbitMQ import (
	Connection,
	Configuration,
	Subscription,
	Reject,
)

class Add(
	Run,
	parameters=Subscription(
		queue='add',
	),
	on_error=Reject(),
):
	def run(
		self,
		a: int,
		b: int,
	):
		print(a + b)

class DeclareQueues(WorkerSetup):
	def __call__(
		self,
		worker: Worker,
	) -> None:
		broker = worker.broker
		if not isinstance(
			broker,
			Connection,
		):
			raise TypeError('A RabbitMQ broker is required')
		for task in worker.tasks:
			declare(
				task=task,
				broker=broker,
			)

broker = Connection(Configuration(
	host='127.0.0.1',
	port=5672,
	username='guest',
	password='guest',
))
worker = Worker(
	broker,
	ThreadPool(10),
	setups=DeclareQueues(),
)
worker.add(Add)
worker.run()

Provision queues in WorkerSetup or separately before starting. Worker connects the injected broker, runs its setup hooks, creates its consumer, and subscribes only registered Runs. It owns that broker connection and pool for one run and closes both on exit; do not share them with another Worker. Call worker.stop() to stop accepting work, finish outstanding tasks and message confirmations, then close resources. Shutdown currently waits for running tasks without a timeout. Pool.submit(RunClass, body) replaces event-name-based submission. It accepts callback(result) and error_callback(exception) keyword arguments, like apply_async, while retaining the executor implementation and returned Future. Only one outcome callback runs, after the Run and its hooks finish. These callbacks stay in the parent process, including with ProcessPool; they must schedule broker operations through Consumer.call, not execute them directly. Worker uses these callbacks for ACK and the Run's failure policy. Both Pool.stop() and Pool.shutdown() wait for tasks; forced termination is not provided by this executor-based implementation. Worker is not an event handler or callable. It creates the consumer without a default handler and supplies a Run-specific handler for each subscription. For standalone broker consumers, consumer(handler) and subs(queue) remain available for standalone consumption. There is one subscription method: subs(queue_or_settings, handler=None). It accepts a queue name, a RabbitMQ Subscription, or a settings dict. An explicit handler overrides the consumer's default handler; one of them is required. Queue-name calls retain the consumer's default offset and QoS. The separate subscribe method has been removed.

RabbitMQ subscription settings can be a Subscription(queue=..., offset=..., qos=..., error=None)

error optionally records a non-empty dead-letter queue name. Subscription itself does not provision resources. Explicit declare(task=task, broker=connection) (or connection.declare(task.parameters)) creates the DLQ and configures the source queue to dead-letter to it. consumer.subs() still never declares resources. Reject sends to the DLQ; Requeue does not. Existing queues with conflicting declaration arguments raise the broker error rather than being silently changed. or a dict with those fields. Unknown settings are rejected, not ignored. Set qos in each Run's subscription (mapped to RabbitMQ's prefetch_count) to limit its unacknowledged deliveries independently (global_qos=False). Omitted/None uses the Consumer's qos default (1); 0 means unlimited, and valid values are 0 through 65535. The limit includes execution and hooks until ACK or failure settlement. It is not a per-Run execution pool size: Runs share the Worker pool. Redis Streams COUNT and Kafka max.poll.records are batch-return limits, not equivalent broker-enforced unacknowledged-message limits; they are not exposed as a universal Run QoS setting. Duplicate queue registrations are rejected: this version associates one queue with one Run, not message-type-based routing. queue.error is not implemented; configure dead-letter routing on the broker separately.

Each delivery creates a Run instance and calls it with instance(**body); Run.__call__ manages hooks and delegates to run. Bodies must be mappings with string keys. Pool capacity includes submitted/running tasks and has no unbounded waiting queue. The default pool size is the CPU count (or 1). A full pool causes NACK with requeue=True. Successful tasks are ACKed only after completion. Failures are logged and use the Run's on_error policy: RabbitMQ defaults to Requeue(), or use Reject() for requeue=False. Requeue does not guarantee placement at the tail and may cause rapid repeated delivery. Reject uses the broker's dead-letter routing if configured, otherwise the message is discarded. Delivery is at-least-once; handlers must tolerate duplicates. NoAck is not a failure policy.

Run classes also accept setups, completes, and errors, each as a single hook or an ordered sequence. The RunSetup, RunComplete, RunError, and Context interfaces are exported from DefStream.Worker. errors contains execution-error hooks (RunError), while on_error selects the failed-event policy (EventError, exported from DefStream.Broker). These are separate responsibilities: error hooks run in the pool, then the event policy runs on the connection owner thread. The former FailurePolicy and error_policy names have been replaced by EventError and on_error. RunSetup hooks receive Context (args and kwargs), completion hooks receive that context and the result, and error hooks receive context and the exception. Setups run before run; completes run after success; errors run after run failure. RunError hooks do not turn failure into success. If any hook raises, remaining hooks stop immediately and the failure policy applies (no recursive error-hook execution). Run instances are constructed at submission, after capacity is acquired. Constructor exceptions are submission errors, not execution errors. The hook/execution lifecycle runs in the pool, and only then is ACK or the failure policy applied on the connection thread. Pool capacity remains occupied until hooks finish. Hooks used with ProcessPool must also be picklable.

Lifecycle hooks

All hook interfaces are exported from DefStream.Worker:

Scope	Hooks	Arguments
Run	RunSetup, RunComplete, RunError	Context; context and result; context and exception
Pool	PoolSetup, PoolCleanup	No arguments; executed inside each pool thread/process
Worker	WorkerSetup, WorkerCleanup	The Worker instance

Runner has been renamed to Run, in the DefStream.Worker.Run module. The old Setup, Complete, Error, and Initializer names have been replaced by the scoped names above; there are no compatibility aliases.

Pass setups= and cleanups= to Worker, ThreadPool, or ProcessPool. Each accepts a single hook or an ordered sequence. Worker setup hooks run after broker connection and before subscriptions; worker.tasks contains the registered Run classes. Use these hooks to declare queues/DLQs or check application dependencies. Worker cleanup hooks run after the pool has shut down and before the consumer and broker are closed, including when connection, setup, subscription, or consumption fails. They must tolerate partially initialized resources.

Pool execution environments start lazily on the first submission. Each runs its setup hooks once, processes tasks, and runs its cleanup hooks in the same thread/process on graceful shutdown. An unused pool runs neither hook. Cleanup runs only after all setup hooks in that environment succeed, including when a task fails; a failing setup must release its own partially acquired resources. Initialize actual task DB connections here, not in the parent process. ProcessPool uses spawned processes; hooks and their configuration must be picklable. Pool setup and cleanup hooks do not receive the Worker or broker.

Persistent execution loops run on the standard executors so that cleanup executes in finally inside the owning environment, rather than submitting arbitrary cleanup tasks that might all land on one worker. Futures, bounded capacity, and parent-side completion callbacks are retained. A ProcessPool also owns a management process for task/result transport, closed during shutdown.

Cleanup hooks run in registration order. A cleanup failure does not skip later cleanup hooks or resource closure; failures propagate with exception chaining. Pool environment failures fail outstanding Futures and are also raised by stop()/shutdown(). Ordinary Run failures do not break the pool. Cleanup cannot be guaranteed after forced termination or a process crash.

Broker confirmations run on the connection owner thread, not pool threads. DefStream.Broker.Event is the common event contract (body, ack, nack, reject); RabbitMQ's Event implements it and retains its broker metadata. Worker and failure policies use this contract; there is no separate Delivery. Only the Run instance and body are sent to ProcessPool children; Run classes must be importable at module scope and instances/inputs/results must be picklable. Use the if __name__ == '__main__': guard with ProcessPool. For graceful Ctrl+C shutdown, register a parent-side SIGINT handler that calls worker.stop() (as in the RabbitMQ sample). Pool child processes ignore SIGINT so they can finish active work and run PoolCleanup instead of raising KeyboardInterrupt. The parent drains message confirmations, stops the pool, runs WorkerCleanup, and closes broker resources. The library does not replace the application's parent-side signal handler.

Install the core and both adapters before running the test suite. Tests use the installed DefStream package rather than importing the source directory as src. Run unit tests with python -m unittest discover -s test -p "test_*.py". The core suite retains shared Worker, Inspector, and cross-adapter contract tests. Adapter-specific unit and integration tests live in each adapter's test directory. From the common parent directory, run all three suites:

python -m unittest discover -s .\DefStream\test -p "test_*.py"
python -m unittest discover -s .\DefStream-RabbitMQ\test -p "test_*.py"
python -m unittest discover -s .\DefStream-Redis\test -p "test_*.py"

To include local RabbitMQ integration tests, set DEFSTREAM_RABBITMQ_TESTS=1 first. They use 127.0.0.1:5672 with guest credentials and create/delete only uniquely named test queues. Redis integration tests are enabled separately with DEFSTREAM_REDIS_TESTS=1.

Examples are organized under:

• sample/Worker/Sample.py for a self-contained in-memory broker and worker. It declares a queue and publishes two messages in WorkerSetup, processes them with a ThreadPool, confirms deliveries on the consumer thread, then exits automatically after draining the queue and runs WorkerCleanup. No RabbitMQ server is required. This finite sample consumes one message at a time and rejects failed messages by default; it is not a persistent broker.
• sample/Worker/RabbitMQ for RabbitMQ-backed workers
• sample/Broker/Implements/RabbitMQ for RabbitMQ queues, exchanges, and streams
• sample/Broker for a small in-memory broker

RabbitMQ examples require a reachable RabbitMQ server and pre-existing queues unless a Worker setup provisions them.

Development in VS Code

The core sources remain in src/Worker and src/Broker. RabbitMQ and Redis implementation files live directly under src in the sibling DefStream-RabbitMQ and DefStream-Redis projects. Their distributions provide only their adapter packages, not the parent packages owned by the core.

From the common parent directory, install all three projects into the workspace virtual environment:

.\.venv\Scripts\python.exe -m pip install `
  --config-settings editable_mode=strict -e .\DefStream
.\.venv\Scripts\python.exe -m pip install `
  -e .\DefStream-RabbitMQ `
  -e .\DefStream-Redis

Select that virtual environment as VS Code's Python interpreter. No manual site-packages/DefStream symbolic link or junction is needed. Remove an existing link before installing the split projects: otherwise installing adapter files under that link can write into the core source directory.

Use strict editable mode for the core and default editable mode for the adapters. The default adapter finders provide parent namespace entries, which can otherwise mask the core's default editable finder. Strict mode exposes the core as regular packages through a link tree under its build directory. Keep that link tree while using the editable installation.

Edits to existing source files are reflected without reinstalling. Reinstall the core after adding or removing its modules. Reinstall the relevant project after changing package metadata or dependencies, recreating the virtual environment, or moving a project directory.