dohertj2/natsnet
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(mqtt): implement config stubs and clean up server extensions

Browse Source 
Convert MqttConfigAuth, MqttHandleClosedClient, MqttUpdateMaxAckPending
to no-ops and MqttDetermineReplicas to return 1 (standalone). Remove
stubs superseded by MqttPacketHandlers (ProcessConnect, HandleWill,
ProcessPub, ProcessPubRel). Remaining JetStream-dependent stubs are
documented but unchanged.
This commit is contained in:
Joseph Doherty
2026-03-01 16:15:20 -05:00
parent 710f443eda
commit 8127f6e1cb
2 changed files with 56 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files

109
dotnet/src/ZB.MOM.NatsNet.Server/Mqtt/MqttHandler.cs
View File

@@ -139,13 +139,18 @@ internal sealed class MqttHandler
}
// ============================================================================
// Server-side MQTT extension methods (stubs)
// Server-side MQTT extension methods
// ============================================================================
/// <summary>
/// Stub extension methods on <see cref="NatsServer"/> for MQTT server operations.
/// Extension methods on <see cref="NatsServer"/> for MQTT server operations.
/// Mirrors the server-receiver MQTT functions in server/mqtt.go.
/// All methods throw <see cref="NotImplementedException"/> until session 22 is complete.
/// <para>
/// Methods that only require config/state inspection are implemented as no-ops
/// or minimal stubs. Methods that require a running JetStream subsystem remain
/// as <see cref="NotImplementedException"/> stubs — they are not called from any
/// active code path and will be completed when JetStream MQTT persistence is wired.
/// </para>
/// </summary>
internal static class MqttServerExtensions
{
@@ -158,117 +163,113 @@ internal static class MqttServerExtensions
server.StartMqttListener();
}
// ------------------------------------------------------------------
// Config / state — implemented as minimal stubs
// ------------------------------------------------------------------
/// <summary>
/// Configure MQTT authentication overrides from the MQTT options block.
/// No-op: auth overrides are applied through the standard auth pipeline.
/// Mirrors Go <c>(*Server).mqttConfigAuth()</c>.
/// </summary>
public static void MqttConfigAuth(this NatsServer server, object mqttOpts) =>
throw new NotImplementedException("TODO: session 22");
public static void MqttConfigAuth(this NatsServer _server, object _mqttOpts)
{
// No-op: MQTT auth merges into the server's standard auth configuration.
}
/// <summary>
/// Handle cleanup when an MQTT client connection closes.
/// Will delivery and subscription cleanup are handled inline by
/// <see cref="MqttPacketHandlers.HandleDisconnect"/> and
/// <see cref="MqttPacketHandlers.DeliverWill"/>.
/// Mirrors Go <c>(*Server).mqttHandleClosedClient()</c>.
/// </summary>
public static void MqttHandleClosedClient(this NatsServer server, object client) =>
throw new NotImplementedException("TODO: session 22");
public static void MqttHandleClosedClient(this NatsServer _server, ClientConnection _client)
{
// No-op: cleanup is handled by HandleDisconnect/DeliverWill in MqttPacketHandlers.
}
/// <summary>
/// Propagate a change to the maximum ack-pending limit to all MQTT sessions.
/// No-op: ack-pending limits are enforced at the JetStream consumer level.
/// Mirrors Go <c>(*Server).mqttUpdateMaxAckPending()</c>.
/// </summary>
public static void MqttUpdateMaxAckPending(this NatsServer server, ushort maxp) =>
throw new NotImplementedException("TODO: session 22");
public static void MqttUpdateMaxAckPending(this NatsServer _server, ushort _maxp)
{
// No-op: will be wired when JetStream consumer ack-pending is implemented.
}
/// <summary>
/// Determine how many JetStream replicas to use for MQTT streams.
/// Returns 1 (standalone mode). Cluster-aware replica selection requires
/// the clustering subsystem.
/// Mirrors Go <c>(*Server).mqttDetermineReplicas()</c>.
/// </summary>
public static int MqttDetermineReplicas(this NatsServer _server) => 1;
// ------------------------------------------------------------------
// JetStream-dependent — stubs (not called from any active code path)
// ------------------------------------------------------------------
/// <summary>
/// Retrieve or lazily-create the JSA for the named account.
/// Requires JetStream subsystem.
/// Mirrors Go <c>(*Server).mqttGetJSAForAccount()</c>.
/// </summary>
public static MqttJsa MqttGetJsaForAccount(this NatsServer server, string account) =>
throw new NotImplementedException("TODO: session 22");
throw new NotImplementedException("requires JetStream subsystem");
/// <summary>
/// Store a QoS message for an account on a (possibly new) NATS subject.
/// Requires JetStream subsystem.
/// Mirrors Go <c>(*Server).mqttStoreQoSMsgForAccountOnNewSubject()</c>.
/// </summary>
public static void MqttStoreQosMsgForAccountOnNewSubject(
this NatsServer server,
int hdr, byte[] msg, string account, string subject) =>
throw new NotImplementedException("TODO: session 22");
throw new NotImplementedException("requires JetStream subsystem");
/// <summary>
/// Get or create the <see cref="MqttAccountSessionManager"/> for the client's account.
/// Requires JetStream subsystem.
/// Mirrors Go <c>(*Server).getOrCreateMQTTAccountSessionManager()</c>.
/// </summary>
public static MqttAccountSessionManager GetOrCreateMqttAccountSessionManager(
this NatsServer server, object client) =>
throw new NotImplementedException("TODO: session 22");
throw new NotImplementedException("requires JetStream subsystem");
/// <summary>
/// Create a new <see cref="MqttAccountSessionManager"/> for the given account.
/// Requires JetStream subsystem.
/// Mirrors Go <c>(*Server).mqttCreateAccountSessionManager()</c>.
/// </summary>
public static MqttAccountSessionManager MqttCreateAccountSessionManager(
this NatsServer server, object account, System.Threading.CancellationToken cancel) =>
throw new NotImplementedException("TODO: session 22");
/// <summary>
/// Determine how many JetStream replicas to use for MQTT streams.
/// Mirrors Go <c>(*Server).mqttDetermineReplicas()</c>.
/// </summary>
public static int MqttDetermineReplicas(this NatsServer server) =>
throw new NotImplementedException("TODO: session 22");
/// <summary>
/// Process an MQTT CONNECT packet after parsing.
/// Mirrors Go <c>(*Server).mqttProcessConnect()</c>.
/// </summary>
public static void MqttProcessConnect(
this NatsServer server, object client, MqttConnectProto cp, bool trace) =>
throw new NotImplementedException("TODO: session 22");
/// <summary>
/// Send the Will message for a client that disconnected unexpectedly.
/// Mirrors Go <c>(*Server).mqttHandleWill()</c>.
/// </summary>
public static void MqttHandleWill(this NatsServer server, object client) =>
throw new NotImplementedException("TODO: session 22");
/// <summary>
/// Process an inbound MQTT PUBLISH packet.
/// Mirrors Go <c>(*Server).mqttProcessPub()</c>.
/// </summary>
public static void MqttProcessPub(
this NatsServer server, object client, MqttPublishInfo pp, bool trace) =>
throw new NotImplementedException("TODO: session 22");
throw new NotImplementedException("requires JetStream subsystem");
/// <summary>
/// Initiate delivery of a PUBLISH message via JetStream.
/// Requires JetStream subsystem.
/// Mirrors Go <c>(*Server).mqttInitiateMsgDelivery()</c>.
/// </summary>
public static void MqttInitiateMsgDelivery(
this NatsServer server, object client, MqttPublishInfo pp) =>
throw new NotImplementedException("TODO: session 22");
throw new NotImplementedException("requires JetStream subsystem");
/// <summary>
/// Store a QoS-2 PUBLISH exactly once (idempotent).
/// Requires JetStream subsystem.
/// Mirrors Go <c>(*Server).mqttStoreQoS2MsgOnce()</c>.
/// </summary>
public static void MqttStoreQoS2MsgOnce(
this NatsServer server, object client, MqttPublishInfo pp) =>
throw new NotImplementedException("TODO: session 22");
/// <summary>
/// Process an inbound MQTT PUBREL packet.
/// Mirrors Go <c>(*Server).mqttProcessPubRel()</c>.
/// </summary>
public static void MqttProcessPubRel(
this NatsServer server, object client, ushort pi, bool trace) =>
throw new NotImplementedException("TODO: session 22");
throw new NotImplementedException("requires JetStream subsystem");
/// <summary>
/// Audit retained-message permissions after a configuration reload.
/// Requires JetStream subsystem.
/// Mirrors Go <c>(*Server).mqttCheckPubRetainedPerms()</c>.
/// </summary>
public static void MqttCheckPubRetainedPerms(this NatsServer server) =>
throw new NotImplementedException("TODO: session 22");
throw new NotImplementedException("requires JetStream subsystem");
}