Zongsoft.Messaging.ZeroMQ 2.4.2

Zongsoft.Messaging.ZeroMQ Message Queue Plugin

English | 简体中文

<a name="abstract"></a>

Abstract

Zongsoft.Messaging.ZeroMQ is a NetMQ-based adapter for the messaging and communication abstractions in Zongsoft.Core. It provides topic publishing and subscription through IMessageQueue, and also supplies request/response and event-channel adapters.

The included ZeroQueueServer combines an XPUB/XSUB exchange for at-most-once traffic with a durable acknowledgement path for at-least-once traffic. Clients discover the current Broker epoch and all runtime endpoints automatically.

<a name="features"></a>

Features

• Implements the Zongsoft IMessageQueue, IRequester, IResponder, and IEventChannel abstractions;
• Supports multiple publishers and subscribers through an XPUB/XSUB exchange;
• Supports topic prefixes, optional message groups, instance filtering, and heartbeats;
• Supports Brotli, GZip, ZLib, or Deflate payload compression above a configurable threshold;
• Supports immediate at-most-once broadcast and Broker-persisted, explicitly acknowledged competing at-least-once delivery;
• Supports standalone use and Zongsoft plugin-based hosting;
• Targets .NET 8, .NET 9, and .NET 10.

<a name="installation"></a>

Installation

Install the NuGet package:

dotnet add package Zongsoft.Messaging.ZeroMQ

To build from this repository, build Zongsoft.Core first:

dotnet build Zongsoft.Core/src/Zongsoft.Core.csproj
dotnet build messaging/zero/Zongsoft.Messaging.ZeroMQ.slnx

<a name="topology"></a>

Exchange Topology

The server uses explicit fixed addresses. A directly constructed server defaults to Discovery 7969, Incoming 32101, and Outgoing 32102, with Control and Storage unset. The daemon configuration explicitly enables Control 32100. Clients rediscover runtime ports after a stopped server changes addresses and restarts.

The server binds each role to its configured IP and does not configure authentication or encryption. NetMQ IPv6 mode may also accept IPv4 through the [::] wildcard; it does not promise IPv6-only wildcard binding. Restrict access at the host or network boundary, or add an authenticated transport before using it across an untrusted network.

<a name="configuration"></a>

Configuration

Server

The daemon option file supplies both server settings and binding declarations:

<?xml version="1.0" encoding="utf-8" ?>

<?zongsoft.tools.binding name="zeromq/control"
	path="Messaging/ZeroMQ/Server"
	filter="enabled:true"
	template="${option.control}"
	protocols="tcp"?>
<?zongsoft.tools.binding name="zeromq/incoming"
	path="Messaging/ZeroMQ/Server"
	filter="enabled:true"
	template="${option.incoming}"
	protocols="tcp"?>
<?zongsoft.tools.binding name="zeromq/outgoing"
	path="Messaging/ZeroMQ/Server"
	filter="enabled:true"
	template="${option.outgoing}"
	protocols="tcp"?>
<?zongsoft.tools.binding name="zeromq/discovery"
	path="Messaging/ZeroMQ/Server"
	filter="enabled:true"
	template="${option.discovery}"
	protocols="tcp"?>

<options>
	<option path="/Messaging/ZeroMQ">
		<server name="QueueServer" enabled="true"
		        storage="QueueServer"
		        control="tcp://0.0.0.0:32100"
		        incoming="tcp://0.0.0.0:32101"
		        outgoing="tcp://0.0.0.0:32102"
		        discovery="tcp://0.0.0.0:7969" />
	</option>
</options>

Discovery, Incoming, and Outgoing are required. Configure Control and Storage together to enable reliable delivery; the Storages factory must create the named storage. Missing dependencies or failed creation prevent startup. Omit or empty both fields for broadcast only. Name identifies the broker, while Storage selects an independent storage configuration. Enabled=false prevents startup and excludes all four bindings; an empty Control produces no binding.

Addresses require tcp, a literal IP address or IPv4 wildcard *, and an explicit port from 1 through 65535. Bracket IPv6 addresses. DNS names, non-root paths, queries, fragments, and random ports are rejected. Each role must use a distinct port. Addresses and storage settings can change only while stopped; startup arguments do not override them. The protocols="tcp" values are constant templates describing the TCP transport.

Discovery advertises ports only; clients reuse the discovery host. All roles must therefore be reachable through that host at the same advertised ports. Separate listening IPs do not advertise separate hosts or support arbitrary NAT port translation.

The standalone server sample starts a loopback broadcast server without persistence:

using var server = new ZeroQueueServer()
{
	Discovery = "tcp://127.0.0.1:7969",
	Incoming = "tcp://127.0.0.1:32101",
	Outgoing = "tcp://127.0.0.1:32102",
};
await server.StartAsync([]);

Client Connection

Define a ZeroMQ connection under /Messaging/ConnectionSettings:

The client name and group come from the sample client, expressed here in host option format:

<configuration>
	<option path="/Messaging">
		<connectionSettings default="ZeroMQ">
			<connectionSetting connectionSetting.name="ZeroMQ"
			                   driver="ZeroMQ"
			                   value="server=127.0.0.1;port=7969;group=Demo;client=Zongsoft.Messaging.ZeroMQ.Sample;" />
		</connectionSettings>
	</option>
</configuration>

The default filter excludes messages produced by the same queue instance. Use Filter=* to accept every instance, Filter=. (or ~) to accept only the current instance, ordinary identifiers as an allow list, and !identifier entries as exclusions.

ZeroConnectionSettings has no TOptions constraint: unknown text entries remain in the main dictionary, and the indexer accepts arbitrary parameter names. The driver reads the parameters it uses through typed properties.

The defaults in the table apply to typed reads. For example, when Port is absent, settings.Port returns 7969, while settings["Port"] returns null; the default is not inserted into Value. Value retains main parameters and explicit extension properties. See the Core settings contract for the complete rules.

<a name="usage"></a>

Usage

Publish and Subscribe

In a host with the ZeroMQ plugin deployed, obtain the shared queue by provider and connection name. This uses the ZeroMQ connection above; for this in-process loopback example set filter=* on that connection and start a compatible broker beforehand. The consumer module references only Core:

using System.Text;
using Zongsoft.Services;
using Zongsoft.Messaging;

var provider = ApplicationContext.Current.Services
	.FindRequired<IMessageQueueProvider>("ZeroMQ");
var queue = provider.Queue("ZeroMQ");
var received = new TaskCompletionSource<string>(TaskCreationOptions.RunContinuationsAsynchronously);
var topic = $"docs/{Guid.NewGuid():N}";

var consumer = await queue.SubscribeAsync(topic, message =>
	received.TrySetResult(Encoding.UTF8.GetString(message.Data.Span)));

try
{
	var identifier = await queue.ProduceAsync(topic, "demo".AsMemory());
	if(identifier == null)
		throw new InvalidOperationException("No matching subscription was visible.");

	Console.WriteLine(await received.Task.WaitAsync(TimeSpan.FromSeconds(10)));
}
finally
{
	await consumer.UnsubscribeAsync();
}

The provider reuses named queues; business operations must not dispose a shared queue. This example cancels only its own unique-topic subscription and waits for its handler before exiting. A non-null publish identifier still does not mean business processing completed. A standalone tool that constructs ZeroQueue directly owns its entire lifetime.

Subscriptions use prefix matching. One ZeroQueue keeps one consumer for each logical topic; subscribing to the same topic again returns the existing consumer and does not replace its handler or options. With Group=Demo, the physical wire topic is Demo:topic/reliable, while handlers receive the logical Message.Topic value topic/reliable.

Each subscriber invokes its handler sequentially in receive order. When its bounded pending queue reaches capacity, that subscriber pauses Poller reads and resumes after the handler frees capacity; other sockets remain responsive.

Compression

Set MessageEnqueueOptions.Compression to an algorithm and the minimum payload size that enables compression. MessageCompression.Value is an integer byte threshold; its text format is <algorithm>:<threshold>. Zero compresses every non-empty payload and the default value disables compression:

var options = new MessageEnqueueOptions()
{
	Compression = MessageCompression.Parse("Brotli:4096"),
};

await queue.ProduceAsync("documents/updated", payload, options);

The equivalent strongly typed construction is new MessageCompression("Brotli", 4096).

At-least-once delivery

Set LeastOnce on both the subscription and publication. Only the Broker requires an IMessageStorage; publishers do not persist messages. A handler must call AcknowledgeAsync; returning normally is not an acknowledgement.

The following is from ZeroQueueReliabilityTests.LeastOnceCompletesAfterBrokerAcceptanceBeforeAcknowledge. ReliableServerScope, CreateQueue and AcknowledgingHandler are real fixtures in that test file. Read or run them in the original test project; they are not application plugin types:

await using var scope = await ReliableServerScope.StartAsync();
using var publisher = CreateQueue(scope.Port, "publisher", "publisher");
using var subscriber = CreateQueue(scope.Port, "subscriber", "subscriber");
var handler = new AcknowledgingHandler(1, false);
await subscriber.SubscribeAsync("topic/reliable", handler, ReliableSubscribeOptions());

var identifier = await publisher.ProduceAsync("topic/reliable", Encoding.UTF8.GetBytes("reliable"), ReliableEnqueueOptions()).AsTask().WaitAsync(TimeSpan.FromSeconds(5));
var message = await handler.ReceiveAsync(TimeSpan.FromSeconds(5));

Assert.False(string.IsNullOrWhiteSpace(identifier));
Assert.Equal(identifier, message.Identifier);
Assert.Single(await GetPendingAsync(scope));

await message.AcknowledgeAsync();
Assert.True(await WaitForPendingCountAsync(scope, 0, TimeSpan.FromSeconds(5)));

The fixture checks that Pending remains before acknowledgement and is removed afterwards. It is not an order-storage implementation or proof of idempotent business effects. Its in-process test storage does not establish production restart durability.

The Broker accepts a publication only when an online matching subscription exists. No match returns null without writing Storage. With a match, the Broker persists Pending first and then returns the identifier. Delivery competes among online subscribers; any one acknowledgement removes Pending. Retries reuse Message.Identifier and may choose another consumer, so handlers must be idempotent.

Set ZeroQueueServer.Storage and Storages only while stopped. With Control enabled, first start calls Storages.Create(Storage) and the server owns the result. Stop retains it for restart; changing the storage name or factory, a failed start, or server disposal releases it, preferring IAsyncDisposable. When Control and Storage are unset, no factory is called and discovery includes only Incoming and Outgoing; LeastOnce is unavailable.

Request and Response

ZeroRequester and ZeroResponder exchange requests through logical URL topics; the default reply topic is <url>/reply. The real ZeroRequesterTests.RequesterReceivesImmediateResponses starts separate requester/responder queues, registers the EchoHandler fixture, sends to rpc/echo, and checks payload and request-identifier correlation. It stops/disposes the responder in finally and also disposes each request token.

One request can receive multiple replies. Each reply refreshes the ten-minute pending-cache expiry; a Replaced notification preserves the active token. Actual expiration removes its registration, while disposing the token also clears buffered replies. Dispose tokens when finished collecting replies. ZeroRequesterMultipleReplyTests covers 3/8 replies over real sockets and deterministic cache replacement, expiration, late replies, and token/requester disposal. Set ZONGSOFT_MESSAGING_TESTS=true to enable the socket tests.

ZeroResponderTests additionally checks subscription rollback on startup failure. No nonexistent PingHandler is supplied here. Applications use shared communication contracts and host composition; standalone tests own their directly constructed instances.

Event Channel

ZeroQueueEventChannel connects an EventExchanger to queue topics under Events/...:

await using var channel = new ZeroQueueEventChannel(queue);
await channel.OpenAsync(exchanger);
await channel.SendAsync(eventContext);

The plugin manifest registers this channel automatically for hosted applications. Request/response and event channels support Group; the prefix is applied only at the network boundary and adapters always use logical topics.

<a name="semantics"></a>

Delivery Semantics

The selected MessageReliability determines the contract:

• MostOnce is transient broadcast. If the application XPUB sees no matching subscription at send time, ProduceAsync returns null without sending. Otherwise it sends locally once and returns a unique identifier. Every broadcast recipient sees that identifier, but it is not remote or Handler acknowledgement.
• LeastOnce returns null without persistence when no online matching subscription exists. Otherwise the Broker persists Pending first and immediately returns the unique identifier without waiting for a Handler acknowledgement.
• The Broker selects one online consumer per attempt. It retries the same identifier until any valid acknowledgement removes Pending. If all consumers disconnect after acceptance, Pending remains until a subscription returns.
• LeastOnce permits duplicate handler invocations. It does not deduplicate business effects and does not provide exactly-once delivery.
• A Control timeout, caller cancellation, or disconnect may leave the publisher unable to determine whether the Broker accepted the message. These stop only local waiting and cannot revoke acceptance already in progress; a business retry can still produce a duplicate.
• Expired reliable messages are removed from Broker Pending with a diagnostic record.
• A queue snapshots its connection, ports, group, filter, timeout, and heartbeat settings at construction; mutating the original settings object does not reconfigure a running queue;
• Empty business payloads are supported.

Message storage is an independent plugin concept, not part of the ZeroMQ driver. This package provides no default file store. Applications using LeastOnce inject an IMessageStorageFactory; the factory selects the connection named by Server.Storage and creates its exclusive storage. Storage supports exact-topic reads and clears. An implementation must hold a message snapshot before SetAsync returns and provide the required restart durability.

See the ZeroMQ 1.0 protocol for the complete Discovery, Broadcast, and Control frame definitions.

<a name="samples"></a>

Samples and Troubleshooting

The .NET 10 samples contain an interactive exchange server and client. Start the server first, then run one client as a subscriber and another as a publisher. See the sample guide for commands.

If messages are not received:

1. Verify that discovery, control, incoming, and outgoing ports are reachable in the required directions;
2. Verify that publisher and subscriber use the same Group and compatible topic prefixes;
3. Check the Filter setting—self-produced messages are excluded by default;
4. If ProduceAsync returns null, verify that a matching subscription was visible to the Broker at that instant and apply the application's retry policy if appropriate;
5. For LeastOnce, verify Server Storage, the Control endpoint, explicit acknowledgement, and expiration;
6. Inspect Broker Pending data, online subscriptions, and consumer idempotency when reliable delivery remains unresolved.

Plugin-Based Integration

Compose this feature through the host; a package reference supplies compile-time APIs, while plugin loading also requires deployed manifests and runtime assets. See the complete plugin workflow.

The deployment manifest selects the daemon plugin, matching option file, and Storage-daemon plugin together for site=daemon. Client-only hosts load the main plugin and client settings. The Storage plugin depends on Daemon and attaches the factory at /Workspace/Messaging/Storages/Redis; change that contribution to use another factory. Reliable delivery requires this factory and the connection selected by Server.Storage.

Add this fragment to an existing host .deploy (retain Main and the host’s other base manifests; do not replace the whole file):

[plugins zongsoft messaging zeromq]
nuget:Zongsoft.Messaging.ZeroMQ

Run dotnet deploy against a test deployment as explained in the workflow, with the host's framework, platform, architecture and, where needed, site. Pin compatible versions in real deployments; application dependencies such as databases, caches or commercial runtimes are still separate prerequisites.

Additional artifacts listed by the deployment manifest include Zongsoft.Messaging.ZeroMQ.option, Zongsoft.Messaging.ZeroMQ.plugin, Zongsoft.Messaging.ZeroMQ-${site}.plugin, Zongsoft.Messaging.ZeroMQ-${site}.option, Zongsoft.Messaging.ZeroMQ.Storage-${site}.plugin. Retain assemblies, dependencies and satellite resource directories as well. Restart the host after deployment, check plugin loading and service/driver registration, then verify the workflow above; copied files alone do not prove that the feature is active.