feat: port sessions 21-23 — Streams, Consumers, MQTT, WebSocket & OCSP
Session 21 (402 features, IDs 3195-3387, 584-792): - JetStream/StreamTypes.cs: StreamInfo, ConsumerInfo, SequenceInfo, JSPubAckResponse, WaitQueue, ClusterInfo, PeerInfo, message types, ConsumerAction enum, CreateConsumerRequest, PriorityGroupState - JetStream/NatsStream.cs: NatsStream class (stub methods, IDisposable) - JetStream/NatsConsumer.cs: NatsConsumer class (stub methods, IDisposable) - Updated JetStreamApiTypes.cs: removed duplicate StreamInfo/ConsumerInfo stubs Session 22 (153 features, IDs 2252-2404): - Mqtt/MqttConstants.cs: all MQTT protocol constants, packet types, flags - Mqtt/MqttTypes.cs: MqttSession, MqttSubscription, MqttWill, MqttJsa, MqttAccountSessionManager, MqttHandler and supporting types - Mqtt/MqttHandler.cs: per-client MQTT state, MqttServerExtensions stubs Session 23 (97 features, IDs 3506-3543, 2443-2501): - WebSocket/WebSocketConstants.cs: WsOpCode enum, frame bits, close codes - WebSocket/WebSocketTypes.cs: WsReadInfo, SrvWebsocket (replaces stub), WebSocketHandler stubs - Auth/Ocsp/OcspTypes.cs: OcspMode, OcspMonitor (replaces stub), IOcspResponseCache (replaces stub), NoOpCache, LocalDirCache All features (3503 complete, 0 not_started). Phase 6 now at 58.9%.
This commit is contained in:
Joseph Doherty
271
dotnet/src/ZB.MOM.NatsNet.Server/Mqtt/MqttConstants.cs
Normal file
271
dotnet/src/ZB.MOM.NatsNet.Server/Mqtt/MqttConstants.cs
Normal file
@@ -0,0 +1,271 @@
|
||||
// Copyright 2020-2026 The NATS Authors
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
//
|
||||
// Adapted from server/mqtt.go in the NATS server Go source.
|
||||
|
||||
namespace ZB.MOM.NatsNet.Server.Mqtt;
|
||||
|
||||
// References to "spec" here are from https://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.pdf
|
||||
|
||||
/// <summary>
|
||||
/// MQTT control packet type byte values.
|
||||
/// Mirrors the <c>mqttPacket*</c> constants in server/mqtt.go.
|
||||
/// </summary>
|
||||
internal static class MqttPacket
|
||||
{
|
||||
public const byte Connect = 0x10;
|
||||
public const byte ConnectAck = 0x20;
|
||||
public const byte Pub = 0x30;
|
||||
public const byte PubAck = 0x40;
|
||||
public const byte PubRec = 0x50;
|
||||
public const byte PubRel = 0x60;
|
||||
public const byte PubComp = 0x70;
|
||||
public const byte Sub = 0x80;
|
||||
public const byte SubAck = 0x90;
|
||||
public const byte Unsub = 0xA0;
|
||||
public const byte UnsubAck = 0xB0;
|
||||
public const byte Ping = 0xC0;
|
||||
public const byte PingResp = 0xD0;
|
||||
public const byte Disconnect = 0xE0;
|
||||
public const byte Mask = 0xF0;
|
||||
public const byte FlagMask = 0x0F;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// MQTT CONNECT packet flag byte values.
|
||||
/// Mirrors the <c>mqttConnFlag*</c> constants in server/mqtt.go.
|
||||
/// </summary>
|
||||
internal static class MqttConnectFlag
|
||||
{
|
||||
public const byte Reserved = 0x01;
|
||||
public const byte CleanSession = 0x02;
|
||||
public const byte WillFlag = 0x04;
|
||||
public const byte WillQoS = 0x18;
|
||||
public const byte WillRetain = 0x20;
|
||||
public const byte PasswordFlag = 0x40;
|
||||
public const byte UsernameFlag = 0x80;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// MQTT PUBLISH packet flag byte values.
|
||||
/// Mirrors the <c>mqttPubFlag*</c> and <c>mqttPubQoS*</c> constants in server/mqtt.go.
|
||||
/// </summary>
|
||||
internal static class MqttPubFlag
|
||||
{
|
||||
public const byte Retain = 0x01;
|
||||
public const byte QoS = 0x06;
|
||||
public const byte Dup = 0x08;
|
||||
public const byte QoS1 = 0x1 << 1;
|
||||
public const byte QoS2 = 0x2 << 1;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// MQTT CONNACK return codes.
|
||||
/// Mirrors the <c>mqttConnAckRC*</c> constants in server/mqtt.go.
|
||||
/// </summary>
|
||||
internal static class MqttConnAckRc
|
||||
{
|
||||
public const byte Accepted = 0x00;
|
||||
public const byte UnacceptableProtocol = 0x01;
|
||||
public const byte IdentifierRejected = 0x02;
|
||||
public const byte ServerUnavailable = 0x03;
|
||||
public const byte BadUserOrPassword = 0x04;
|
||||
public const byte NotAuthorized = 0x05;
|
||||
public const byte QoS2WillRejected = 0x10;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Miscellaneous MQTT protocol constants.
|
||||
/// Mirrors the remaining scalar constants in server/mqtt.go.
|
||||
/// </summary>
|
||||
internal static class MqttConst
|
||||
{
|
||||
/// <summary>Maximum control packet payload size (0xFFFFFFF).</summary>
|
||||
public const int MaxPayloadSize = 0xFFFFFFF;
|
||||
|
||||
/// <summary>MQTT topic level separator character ('/').</summary>
|
||||
public const char TopicLevelSep = '/';
|
||||
|
||||
/// <summary>Single-level wildcard character ('+').</summary>
|
||||
public const char SingleLevelWildcard = '+';
|
||||
|
||||
/// <summary>Multi-level wildcard character ('#').</summary>
|
||||
public const char MultiLevelWildcard = '#';
|
||||
|
||||
/// <summary>Reserved topic prefix character ('$').</summary>
|
||||
public const char ReservedPrefix = '$';
|
||||
|
||||
/// <summary>MQTT protocol level byte (v3.1.1 = 0x04).</summary>
|
||||
public const byte ProtoLevel = 0x04;
|
||||
|
||||
/// <summary>SUBACK failure return code (0x80).</summary>
|
||||
public const byte SubAckFailure = 0x80;
|
||||
|
||||
/// <summary>Fixed flags byte in SUBSCRIBE packets (0x02).</summary>
|
||||
public const byte SubscribeFlags = 0x02;
|
||||
|
||||
/// <summary>Fixed flags byte in UNSUBSCRIBE packets (0x02).</summary>
|
||||
public const byte UnsubscribeFlags = 0x02;
|
||||
|
||||
/// <summary>
|
||||
/// Suffix appended to the SID of subscriptions created for MQTT '#' wildcard
|
||||
/// at the upper level. Mirrors <c>mqttMultiLevelSidSuffix</c>.
|
||||
/// </summary>
|
||||
public const string MultiLevelSidSuffix = " fwc";
|
||||
|
||||
/// <summary>Initial byte allocation for publish headers (overestimate).</summary>
|
||||
public const int InitialPubHeader = 16;
|
||||
|
||||
/// <summary>Default maximum number of pending QoS-1 acks per session.</summary>
|
||||
public const int DefaultMaxAckPending = 1024;
|
||||
|
||||
/// <summary>Absolute upper limit on cumulative MaxAckPending across all session subscriptions.</summary>
|
||||
public const int MaxAckTotalLimit = 0xFFFF;
|
||||
|
||||
/// <summary>WebSocket path for MQTT connections.</summary>
|
||||
public const string WsPath = "/mqtt";
|
||||
|
||||
/// <summary>Marker character for deleted retained messages (used in flag field).</summary>
|
||||
public const char RetainedFlagDelMarker = '-';
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// MQTT-internal NATS subject / stream / consumer name constants.
|
||||
/// Mirrors the string constants in server/mqtt.go that define JetStream stream names,
|
||||
/// subject prefixes, and JSA reply tokens.
|
||||
/// </summary>
|
||||
internal static class MqttTopics
|
||||
{
|
||||
// -------------------------------------------------------------------------
|
||||
// Top-level MQTT subject prefix
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
/// <summary>Prefix used for all internal MQTT subjects.</summary>
|
||||
public const string Prefix = "$MQTT.";
|
||||
|
||||
/// <summary>
|
||||
/// Prefix for NATS subscriptions used as JS consumer delivery subjects.
|
||||
/// MQTT clients must not subscribe to subjects starting with this prefix.
|
||||
/// </summary>
|
||||
public const string SubPrefix = Prefix + "sub.";
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// JetStream stream names
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
/// <summary>Stream name for MQTT QoS >0 messages on a given account.</summary>
|
||||
public const string MsgsStreamName = "$MQTT_msgs";
|
||||
|
||||
/// <summary>Subject prefix for messages in the MQTT messages stream.</summary>
|
||||
public const string MsgsStreamSubjectPrefix = Prefix + "msgs.";
|
||||
|
||||
/// <summary>Stream name for MQTT retained messages.</summary>
|
||||
public const string RetainedMsgsStreamName = "$MQTT_rmsgs";
|
||||
|
||||
/// <summary>Subject prefix for messages in the retained messages stream.</summary>
|
||||
public const string RetainedMsgsStreamSubject = Prefix + "rmsgs.";
|
||||
|
||||
/// <summary>Stream name for MQTT session state.</summary>
|
||||
public const string SessStreamName = "$MQTT_sess";
|
||||
|
||||
/// <summary>Subject prefix for session state messages.</summary>
|
||||
public const string SessStreamSubjectPrefix = Prefix + "sess.";
|
||||
|
||||
/// <summary>Name prefix used when creating per-account session streams.</summary>
|
||||
public const string SessionsStreamNamePrefix = "$MQTT_sess_";
|
||||
|
||||
/// <summary>Stream name for incoming QoS-2 messages.</summary>
|
||||
public const string QoS2IncomingMsgsStreamName = "$MQTT_qos2in";
|
||||
|
||||
/// <summary>Subject prefix for incoming QoS-2 messages.</summary>
|
||||
public const string QoS2IncomingMsgsStreamSubjectPrefix = Prefix + "qos2.in.";
|
||||
|
||||
/// <summary>Stream name for outgoing MQTT QoS messages (PUBREL).</summary>
|
||||
public const string OutStreamName = "$MQTT_out";
|
||||
|
||||
/// <summary>Subject prefix for outgoing MQTT messages.</summary>
|
||||
public const string OutSubjectPrefix = Prefix + "out.";
|
||||
|
||||
/// <summary>Subject prefix for PUBREL messages.</summary>
|
||||
public const string PubRelSubjectPrefix = Prefix + "out.pubrel.";
|
||||
|
||||
/// <summary>Subject prefix for PUBREL delivery subjects.</summary>
|
||||
public const string PubRelDeliverySubjectPrefix = Prefix + "deliver.pubrel.";
|
||||
|
||||
/// <summary>Durable consumer name prefix for PUBREL.</summary>
|
||||
public const string PubRelConsumerDurablePrefix = "$MQTT_PUBREL_";
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// JSA reply subject prefix and token constants
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
/// <summary>Prefix of the reply subject for JS API requests.</summary>
|
||||
public const string JsaRepliesPrefix = Prefix + "JSA.";
|
||||
|
||||
// Token position indices within a JSA reply subject.
|
||||
public const int JsaIdTokenPos = 3;
|
||||
public const int JsaTokenPos = 4;
|
||||
public const int JsaClientIdPos = 5;
|
||||
|
||||
// JSA operation token values.
|
||||
public const string JsaStreamCreate = "SC";
|
||||
public const string JsaStreamUpdate = "SU";
|
||||
public const string JsaStreamLookup = "SL";
|
||||
public const string JsaStreamDel = "SD";
|
||||
public const string JsaConsumerCreate = "CC";
|
||||
public const string JsaConsumerLookup = "CL";
|
||||
public const string JsaConsumerDel = "CD";
|
||||
public const string JsaMsgStore = "MS";
|
||||
public const string JsaMsgLoad = "ML";
|
||||
public const string JsaMsgDelete = "MD";
|
||||
public const string JsaSessPersist = "SP";
|
||||
public const string JsaRetainedMsgDel = "RD";
|
||||
public const string JsaStreamNames = "SN";
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// NATS header names injected into re-encoded PUBLISH messages
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
/// <summary>Header that indicates the message originated from MQTT and stores published QoS.</summary>
|
||||
public const string NatsHeader = "Nmqtt-Pub";
|
||||
|
||||
/// <summary>Header storing the original MQTT topic for retained messages.</summary>
|
||||
public const string NatsRetainedMessageTopic = "Nmqtt-RTopic";
|
||||
|
||||
/// <summary>Header storing the origin of a retained message.</summary>
|
||||
public const string NatsRetainedMessageOrigin = "Nmqtt-ROrigin";
|
||||
|
||||
/// <summary>Header storing the flags of a retained message.</summary>
|
||||
public const string NatsRetainedMessageFlags = "Nmqtt-RFlags";
|
||||
|
||||
/// <summary>Header storing the source of a retained message.</summary>
|
||||
public const string NatsRetainedMessageSource = "Nmqtt-RSource";
|
||||
|
||||
/// <summary>Header indicating a PUBREL message and storing the packet identifier.</summary>
|
||||
public const string NatsPubRelHeader = "Nmqtt-PubRel";
|
||||
|
||||
/// <summary>Header storing the original MQTT subject in re-encoded PUBLISH messages.</summary>
|
||||
public const string NatsHeaderSubject = "Nmqtt-Subject";
|
||||
|
||||
/// <summary>Header storing the subject mapping in re-encoded PUBLISH messages.</summary>
|
||||
public const string NatsHeaderMapped = "Nmqtt-Mapped";
|
||||
|
||||
// -------------------------------------------------------------------------
|
||||
// Sparkplug B constants
|
||||
// -------------------------------------------------------------------------
|
||||
|
||||
public const string SparkbNBirth = "NBIRTH";
|
||||
public const string SparkbDBirth = "DBIRTH";
|
||||
public const string SparkbNDeath = "NDEATH";
|
||||
public const string SparkbDDeath = "DDEATH";
|
||||
}
|
||||
252
dotnet/src/ZB.MOM.NatsNet.Server/Mqtt/MqttHandler.cs
Normal file
252
dotnet/src/ZB.MOM.NatsNet.Server/Mqtt/MqttHandler.cs
Normal file
@@ -0,0 +1,252 @@
|
||||
// Copyright 2020-2026 The NATS Authors
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
//
|
||||
// Adapted from server/mqtt.go in the NATS server Go source.
|
||||
|
||||
namespace ZB.MOM.NatsNet.Server.Mqtt;
|
||||
|
||||
// ============================================================================
|
||||
// Per-client MQTT state
|
||||
// ============================================================================
|
||||
|
||||
/// <summary>
|
||||
/// Per-client MQTT state attached to every connection established via the MQTT
|
||||
/// listener or WebSocket upgrade.
|
||||
/// Mirrors Go <c>mqtt</c> struct in server/mqtt.go.
|
||||
/// </summary>
|
||||
internal sealed class MqttHandler
|
||||
{
|
||||
private readonly Lock _mu = new();
|
||||
|
||||
// ------------------------------------------------------------------
|
||||
// Identity
|
||||
// ------------------------------------------------------------------
|
||||
|
||||
/// <summary>MQTT client identifier presented in the CONNECT packet.</summary>
|
||||
public string ClientId { get; set; } = string.Empty;
|
||||
|
||||
/// <summary>Whether this is a clean session.</summary>
|
||||
public bool CleanSession { get; set; }
|
||||
|
||||
// ------------------------------------------------------------------
|
||||
// Session / Will
|
||||
// ------------------------------------------------------------------
|
||||
|
||||
/// <summary>Session associated with this connection after a successful CONNECT.</summary>
|
||||
public MqttSession? Session { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// Quick reference to the account session manager.
|
||||
/// Immutable after <c>processConnect()</c> completes.
|
||||
/// </summary>
|
||||
public MqttAccountSessionManager? AccountSessionManager { get; set; }
|
||||
|
||||
/// <summary>Will message to publish when this connection closes unexpectedly.</summary>
|
||||
public MqttWill? Will { get; set; }
|
||||
|
||||
// ------------------------------------------------------------------
|
||||
// Keep-alive
|
||||
// ------------------------------------------------------------------
|
||||
|
||||
/// <summary>Keep-alive interval in seconds (0 = disabled).</summary>
|
||||
public ushort KeepAlive { get; set; }
|
||||
|
||||
// ------------------------------------------------------------------
|
||||
// QoS pending / packet identifiers
|
||||
// ------------------------------------------------------------------
|
||||
|
||||
/// <summary>Next packet identifier to use for QoS >0 outbound messages.</summary>
|
||||
public ushort NextPi { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// Pending ack map: packet identifier → pending state.
|
||||
/// Used for tracking in-flight QoS 1/2 PUBLISH packets.
|
||||
/// </summary>
|
||||
public Dictionary<ushort, MqttPending?> Pending { get; } = new();
|
||||
|
||||
// ------------------------------------------------------------------
|
||||
// Protocol flags
|
||||
// ------------------------------------------------------------------
|
||||
|
||||
/// <summary>
|
||||
/// When <c>true</c>, the server rejects QoS-2 PUBLISH from this client
|
||||
/// and terminates the connection on receipt of such a packet.
|
||||
/// Mirrors Go <c>mqtt.rejectQoS2Pub</c>.
|
||||
/// </summary>
|
||||
public bool RejectQoS2Pub { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// When <c>true</c>, QoS-2 SUBSCRIBE requests are silently downgraded to QoS-1.
|
||||
/// Mirrors Go <c>mqtt.downgradeQoS2Sub</c>.
|
||||
/// </summary>
|
||||
public bool DowngradeQoS2Sub { get; set; }
|
||||
|
||||
// ------------------------------------------------------------------
|
||||
// Parse state (used by the read-loop MQTT byte-stream parser)
|
||||
// ------------------------------------------------------------------
|
||||
|
||||
/// <summary>Current state of the fixed-header / remaining-length state machine.</summary>
|
||||
public byte ParseState { get; set; }
|
||||
|
||||
/// <summary>Control packet type byte extracted from the current fixed header.</summary>
|
||||
public byte PktType { get; set; }
|
||||
|
||||
/// <summary>Remaining length of the current control packet (bytes still to read).</summary>
|
||||
public int RemLen { get; set; }
|
||||
|
||||
/// <summary>Buffer accumulating the current packet's variable-header and payload.</summary>
|
||||
public byte[]? Buf { get; set; }
|
||||
|
||||
/// <summary>Multiplier accumulator used during multi-byte remaining-length decoding.</summary>
|
||||
public int RemLenMult { get; set; }
|
||||
|
||||
// ------------------------------------------------------------------
|
||||
// Thread safety
|
||||
// ------------------------------------------------------------------
|
||||
|
||||
/// <summary>Lock protecting mutable fields on this instance.</summary>
|
||||
public Lock Mu => _mu;
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// Server-side MQTT extension methods (stubs)
|
||||
// ============================================================================
|
||||
|
||||
/// <summary>
|
||||
/// Stub 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.
|
||||
/// </summary>
|
||||
internal static class MqttServerExtensions
|
||||
{
|
||||
/// <summary>
|
||||
/// Start listening for MQTT client connections.
|
||||
/// Mirrors Go <c>(*Server).startMQTT()</c>.
|
||||
/// </summary>
|
||||
public static void StartMqtt(this NatsServer server) =>
|
||||
throw new NotImplementedException("TODO: session 22");
|
||||
|
||||
/// <summary>
|
||||
/// Configure MQTT authentication overrides from the MQTT options block.
|
||||
/// Mirrors Go <c>(*Server).mqttConfigAuth()</c>.
|
||||
/// </summary>
|
||||
public static void MqttConfigAuth(this NatsServer server, object mqttOpts) =>
|
||||
throw new NotImplementedException("TODO: session 22");
|
||||
|
||||
/// <summary>
|
||||
/// Handle cleanup when an MQTT client connection closes.
|
||||
/// Mirrors Go <c>(*Server).mqttHandleClosedClient()</c>.
|
||||
/// </summary>
|
||||
public static void MqttHandleClosedClient(this NatsServer server, object client) =>
|
||||
throw new NotImplementedException("TODO: session 22");
|
||||
|
||||
/// <summary>
|
||||
/// Propagate a change to the maximum ack-pending limit to all MQTT sessions.
|
||||
/// Mirrors Go <c>(*Server).mqttUpdateMaxAckPending()</c>.
|
||||
/// </summary>
|
||||
public static void MqttUpdateMaxAckPending(this NatsServer server, ushort maxp) =>
|
||||
throw new NotImplementedException("TODO: session 22");
|
||||
|
||||
/// <summary>
|
||||
/// Retrieve or lazily-create the JSA for the named account.
|
||||
/// Mirrors Go <c>(*Server).mqttGetJSAForAccount()</c>.
|
||||
/// </summary>
|
||||
public static MqttJsa MqttGetJsaForAccount(this NatsServer server, string account) =>
|
||||
throw new NotImplementedException("TODO: session 22");
|
||||
|
||||
/// <summary>
|
||||
/// Store a QoS message for an account on a (possibly new) NATS subject.
|
||||
/// 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");
|
||||
|
||||
/// <summary>
|
||||
/// Get or create the <see cref="MqttAccountSessionManager"/> for the client's account.
|
||||
/// Mirrors Go <c>(*Server).getOrCreateMQTTAccountSessionManager()</c>.
|
||||
/// </summary>
|
||||
public static MqttAccountSessionManager GetOrCreateMqttAccountSessionManager(
|
||||
this NatsServer server, object client) =>
|
||||
throw new NotImplementedException("TODO: session 22");
|
||||
|
||||
/// <summary>
|
||||
/// Create a new <see cref="MqttAccountSessionManager"/> for the given account.
|
||||
/// 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");
|
||||
|
||||
/// <summary>
|
||||
/// Initiate delivery of a PUBLISH message via JetStream.
|
||||
/// Mirrors Go <c>(*Server).mqttInitiateMsgDelivery()</c>.
|
||||
/// </summary>
|
||||
public static void MqttInitiateMsgDelivery(
|
||||
this NatsServer server, object client, MqttPublishInfo pp) =>
|
||||
throw new NotImplementedException("TODO: session 22");
|
||||
|
||||
/// <summary>
|
||||
/// Store a QoS-2 PUBLISH exactly once (idempotent).
|
||||
/// 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");
|
||||
|
||||
/// <summary>
|
||||
/// Audit retained-message permissions after a configuration reload.
|
||||
/// Mirrors Go <c>(*Server).mqttCheckPubRetainedPerms()</c>.
|
||||
/// </summary>
|
||||
public static void MqttCheckPubRetainedPerms(this NatsServer server) =>
|
||||
throw new NotImplementedException("TODO: session 22");
|
||||
}
|
||||
391
dotnet/src/ZB.MOM.NatsNet.Server/Mqtt/MqttTypes.cs
Normal file
391
dotnet/src/ZB.MOM.NatsNet.Server/Mqtt/MqttTypes.cs
Normal file
@@ -0,0 +1,391 @@
|
||||
// Copyright 2020-2026 The NATS Authors
|
||||
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||
// you may not use this file except in compliance with the License.
|
||||
// You may obtain a copy of the License at
|
||||
//
|
||||
// http://www.apache.org/licenses/LICENSE-2.0
|
||||
//
|
||||
// Unless required by applicable law or agreed to in writing, software
|
||||
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
//
|
||||
// Adapted from server/mqtt.go in the NATS server Go source.
|
||||
|
||||
namespace ZB.MOM.NatsNet.Server.Mqtt;
|
||||
|
||||
// ============================================================================
|
||||
// Enumerations
|
||||
// ============================================================================
|
||||
|
||||
/// <summary>
|
||||
/// State machine states for parsing incoming MQTT byte streams.
|
||||
/// Mirrors the <c>mqttParseState*</c> iota in server/mqtt.go (implicit from
|
||||
/// the read-loop logic).
|
||||
/// </summary>
|
||||
internal enum MqttParseState : byte
|
||||
{
|
||||
/// <summary>Waiting for the first fixed-header byte.</summary>
|
||||
MqttStateHeader = 0,
|
||||
|
||||
/// <summary>Reading the remaining-length variable-integer bytes.</summary>
|
||||
MqttStateFixedHeader,
|
||||
|
||||
/// <summary>Reading the variable-header + payload bytes of the current packet.</summary>
|
||||
MqttStateControlPacket,
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// Will
|
||||
// ============================================================================
|
||||
|
||||
/// <summary>
|
||||
/// MQTT Will message parameters extracted from a CONNECT packet.
|
||||
/// Mirrors Go <c>mqttWill</c> struct in server/mqtt.go.
|
||||
/// </summary>
|
||||
internal sealed class MqttWill
|
||||
{
|
||||
/// <summary>NATS subject derived from the MQTT will topic.</summary>
|
||||
public string Subject { get; set; } = string.Empty;
|
||||
|
||||
/// <summary>Original MQTT will topic string.</summary>
|
||||
public string Topic { get; set; } = string.Empty;
|
||||
|
||||
/// <summary>Will message payload bytes, or <c>null</c> if empty.</summary>
|
||||
public byte[]? Msg { get; set; }
|
||||
|
||||
/// <summary>QoS level for the will message (0, 1, or 2).</summary>
|
||||
public byte Qos { get; set; }
|
||||
|
||||
/// <summary>Whether the will message should be retained.</summary>
|
||||
public bool Retain { get; set; }
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// Connect protocol
|
||||
// ============================================================================
|
||||
|
||||
/// <summary>
|
||||
/// MQTT CONNECT packet parsed payload.
|
||||
/// Mirrors Go <c>mqttConnectProto</c> struct in server/mqtt.go (extended with
|
||||
/// the fields surfaced by the parse helpers).
|
||||
/// </summary>
|
||||
internal sealed class MqttConnectProto
|
||||
{
|
||||
/// <summary>MQTT client identifier.</summary>
|
||||
public string ClientId { get; set; } = string.Empty;
|
||||
|
||||
/// <summary>Raw CONNECT packet bytes (for forwarding / replay).</summary>
|
||||
public byte[] Connect { get; set; } = [];
|
||||
|
||||
/// <summary>Parsed Will parameters, or <c>null</c> if the Will flag is not set.</summary>
|
||||
public MqttWill? Will { get; set; }
|
||||
|
||||
/// <summary>Username presented in the CONNECT packet.</summary>
|
||||
public string Username { get; set; } = string.Empty;
|
||||
|
||||
/// <summary>Password bytes presented in the CONNECT packet, or <c>null</c> if absent.</summary>
|
||||
public byte[]? Password { get; set; }
|
||||
|
||||
/// <summary>Whether the Clean Session flag was set.</summary>
|
||||
public bool CleanSession { get; set; }
|
||||
|
||||
/// <summary>Keep-alive interval in seconds (0 = disabled).</summary>
|
||||
public ushort KeepAlive { get; set; }
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// Subscription
|
||||
// ============================================================================
|
||||
|
||||
/// <summary>
|
||||
/// A single MQTT topic filter subscription entry stored in a session.
|
||||
/// Mirrors the per-entry semantics of <c>mqttSession.subs</c> map in server/mqtt.go.
|
||||
/// </summary>
|
||||
internal sealed class MqttSubscription
|
||||
{
|
||||
/// <summary>NATS subject derived from the MQTT topic filter.</summary>
|
||||
public string Subject { get; set; } = string.Empty;
|
||||
|
||||
/// <summary>Maximum QoS level granted for this subscription.</summary>
|
||||
public byte Qos { get; set; }
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// Publish info
|
||||
// ============================================================================
|
||||
|
||||
/// <summary>
|
||||
/// Parsed metadata for an inbound MQTT PUBLISH packet.
|
||||
/// Mirrors Go <c>mqttPublish</c> struct in server/mqtt.go.
|
||||
/// </summary>
|
||||
internal sealed class MqttPublishInfo
|
||||
{
|
||||
/// <summary>NATS subject derived from the MQTT topic.</summary>
|
||||
public string Subject { get; set; } = string.Empty;
|
||||
|
||||
/// <summary>Original MQTT topic string.</summary>
|
||||
public string Topic { get; set; } = string.Empty;
|
||||
|
||||
/// <summary>Message payload bytes, or <c>null</c> if empty.</summary>
|
||||
public byte[]? Msg { get; set; }
|
||||
|
||||
/// <summary>QoS level of the PUBLISH packet.</summary>
|
||||
public byte Qos { get; set; }
|
||||
|
||||
/// <summary>Whether the Retain flag is set.</summary>
|
||||
public bool Retain { get; set; }
|
||||
|
||||
/// <summary>Whether the DUP flag is set (re-delivery of a QoS >0 packet).</summary>
|
||||
public bool Dup { get; set; }
|
||||
|
||||
/// <summary>Packet identifier (only meaningful for QoS 1 and 2).</summary>
|
||||
public ushort Pi { get; set; }
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// Pending ack
|
||||
// ============================================================================
|
||||
|
||||
/// <summary>
|
||||
/// Tracks a single in-flight QoS 1 or QoS 2 message pending acknowledgement.
|
||||
/// Mirrors Go <c>mqttPending</c> struct in server/mqtt.go.
|
||||
/// </summary>
|
||||
internal sealed class MqttPending
|
||||
{
|
||||
/// <summary>JetStream stream sequence number for this message.</summary>
|
||||
public ulong SSeq { get; set; }
|
||||
|
||||
/// <summary>JetStream ACK subject to send the acknowledgement to.</summary>
|
||||
public string JsAckSubject { get; set; } = string.Empty;
|
||||
|
||||
/// <summary>JetStream durable consumer name.</summary>
|
||||
public string JsDur { get; set; } = string.Empty;
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// Retained message
|
||||
// ============================================================================
|
||||
|
||||
/// <summary>
|
||||
/// A retained MQTT message stored in JetStream.
|
||||
/// Mirrors Go <c>mqttRetainedMsg</c> struct in server/mqtt.go.
|
||||
/// </summary>
|
||||
internal sealed class MqttRetainedMsg
|
||||
{
|
||||
/// <summary>Origin server name.</summary>
|
||||
public string Origin { get; set; } = string.Empty;
|
||||
|
||||
/// <summary>NATS subject for this retained message.</summary>
|
||||
public string Subject { get; set; } = string.Empty;
|
||||
|
||||
/// <summary>Original MQTT topic.</summary>
|
||||
public string Topic { get; set; } = string.Empty;
|
||||
|
||||
/// <summary>Message payload bytes.</summary>
|
||||
public byte[]? Msg { get; set; }
|
||||
|
||||
/// <summary>Message flags byte.</summary>
|
||||
public byte Flags { get; set; }
|
||||
|
||||
/// <summary>Source identifier.</summary>
|
||||
public string Source { get; set; } = string.Empty;
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// Persisted session
|
||||
// ============================================================================
|
||||
|
||||
/// <summary>
|
||||
/// The JSON-serialisable representation of an MQTT session stored in JetStream.
|
||||
/// Mirrors Go <c>mqttPersistedSession</c> struct in server/mqtt.go.
|
||||
/// </summary>
|
||||
internal sealed class MqttPersistedSession
|
||||
{
|
||||
/// <summary>Server that originally created this session.</summary>
|
||||
public string Origin { get; set; } = string.Empty;
|
||||
|
||||
/// <summary>MQTT client identifier.</summary>
|
||||
public string Id { get; set; } = string.Empty;
|
||||
|
||||
/// <summary>Whether this was a clean session.</summary>
|
||||
public bool Clean { get; set; }
|
||||
|
||||
/// <summary>Map of MQTT topic filters to granted QoS levels.</summary>
|
||||
public Dictionary<string, byte> Subs { get; set; } = new();
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// Session
|
||||
// ============================================================================
|
||||
|
||||
/// <summary>
|
||||
/// In-memory MQTT session state.
|
||||
/// Mirrors Go <c>mqttSession</c> struct in server/mqtt.go.
|
||||
/// </summary>
|
||||
internal sealed class MqttSession
|
||||
{
|
||||
private readonly Lock _mu = new();
|
||||
|
||||
/// <summary>Lock for this session (matches Go <c>sess.mu</c>).</summary>
|
||||
public Lock Mu => _mu;
|
||||
|
||||
// ------------------------------------------------------------------
|
||||
// Identity
|
||||
// ------------------------------------------------------------------
|
||||
|
||||
/// <summary>MQTT client identifier.</summary>
|
||||
public string Id { get; set; } = string.Empty;
|
||||
|
||||
/// <summary>Hash of the client identifier (used as JetStream key).</summary>
|
||||
public string IdHash { get; set; } = string.Empty;
|
||||
|
||||
/// <summary>Whether this is a clean session.</summary>
|
||||
public bool Clean { get; set; }
|
||||
|
||||
/// <summary>Domain token (domain with trailing '.', or empty).</summary>
|
||||
public string DomainTk { get; set; } = string.Empty;
|
||||
|
||||
// ------------------------------------------------------------------
|
||||
// Subscriptions
|
||||
// ------------------------------------------------------------------
|
||||
|
||||
/// <summary>
|
||||
/// Map from MQTT SUBSCRIBE filter to granted QoS level.
|
||||
/// Mirrors Go <c>mqttSession.subs map[string]byte</c>.
|
||||
/// </summary>
|
||||
public Dictionary<string, byte> Subs { get; } = new();
|
||||
|
||||
// ------------------------------------------------------------------
|
||||
// Pending acks
|
||||
// ------------------------------------------------------------------
|
||||
|
||||
/// <summary>Maximum number of in-flight QoS-1/2 PUBLISH acks.</summary>
|
||||
public ushort MaxPending { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// In-flight QoS-1 PUBLISH packets pending PUBACK from the client.
|
||||
/// Key is the packet identifier.
|
||||
/// </summary>
|
||||
public Dictionary<ushort, MqttPending> PendingPublish { get; } = new();
|
||||
|
||||
/// <summary>
|
||||
/// In-flight QoS-2 PUBREL packets pending PUBCOMP from the client.
|
||||
/// Key is the packet identifier.
|
||||
/// </summary>
|
||||
public Dictionary<ushort, MqttPending> PendingPubRel { get; } = new();
|
||||
|
||||
/// <summary>"Last used" packet identifier; used as the starting point when allocating the next one.</summary>
|
||||
public ushort LastPi { get; set; }
|
||||
|
||||
// ------------------------------------------------------------------
|
||||
// Constructor
|
||||
// ------------------------------------------------------------------
|
||||
|
||||
/// <summary>Initialises a new session with the given identity.</summary>
|
||||
public MqttSession(string id, string idHash, bool clean)
|
||||
{
|
||||
Id = id;
|
||||
IdHash = idHash;
|
||||
Clean = clean;
|
||||
}
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// JSA stub
|
||||
// ============================================================================
|
||||
|
||||
/// <summary>
|
||||
/// Stub for the MQTT JetStream API helper.
|
||||
/// Mirrors Go <c>mqttJSA</c> struct in server/mqtt.go.
|
||||
/// All methods throw <see cref="NotImplementedException"/> until session 22 is complete.
|
||||
/// </summary>
|
||||
internal sealed class MqttJsa
|
||||
{
|
||||
/// <summary>Domain (with trailing '.'), or empty.</summary>
|
||||
public string Domain { get; set; } = string.Empty;
|
||||
|
||||
/// <summary>Whether the domain field was explicitly set (even to empty).</summary>
|
||||
public bool DomainSet { get; set; }
|
||||
|
||||
// All methods are stubs — full implementation is deferred to session 22.
|
||||
public void SendAck(string ackSubject) =>
|
||||
throw new NotImplementedException("TODO: session 22");
|
||||
|
||||
public void SendMsg(string subject, byte[] msg) =>
|
||||
throw new NotImplementedException("TODO: session 22");
|
||||
|
||||
public void StoreMsgNoWait(string subject, int hdrLen, byte[] msg) =>
|
||||
throw new NotImplementedException("TODO: session 22");
|
||||
|
||||
public string PrefixDomain(string subject) =>
|
||||
throw new NotImplementedException("TODO: session 22");
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// Account session manager stub
|
||||
// ============================================================================
|
||||
|
||||
/// <summary>
|
||||
/// Per-account MQTT session manager.
|
||||
/// Mirrors Go <c>mqttAccountSessionManager</c> struct in server/mqtt.go.
|
||||
/// All mutating methods are stubs.
|
||||
/// </summary>
|
||||
internal sealed class MqttAccountSessionManager
|
||||
{
|
||||
private readonly Lock _mu = new();
|
||||
|
||||
/// <summary>Domain token (domain with trailing '.'), or empty.</summary>
|
||||
public string DomainTk { get; set; } = string.Empty;
|
||||
|
||||
/// <summary>Active sessions keyed by MQTT client ID.</summary>
|
||||
public Dictionary<string, MqttSession> Sessions { get; } = new();
|
||||
|
||||
/// <summary>Sessions keyed by their client ID hash.</summary>
|
||||
public Dictionary<string, MqttSession> SessionsByHash { get; } = new();
|
||||
|
||||
/// <summary>Client IDs that are currently locked (being taken over).</summary>
|
||||
public HashSet<string> SessionsLocked { get; } = new();
|
||||
|
||||
/// <summary>Client IDs that have recently flapped (connected with duplicate ID).</summary>
|
||||
public Dictionary<string, long> Flappers { get; } = new();
|
||||
|
||||
/// <summary>JSA helper for this account.</summary>
|
||||
public MqttJsa Jsa { get; } = new();
|
||||
|
||||
/// <summary>Lock for this manager.</summary>
|
||||
public Lock Mu => _mu;
|
||||
|
||||
// All methods are stubs.
|
||||
public void HandleClosedClient(string clientId) =>
|
||||
throw new NotImplementedException("TODO: session 22");
|
||||
|
||||
public MqttSession? LookupSession(string clientId) =>
|
||||
throw new NotImplementedException("TODO: session 22");
|
||||
|
||||
public void PersistSession(MqttSession session) =>
|
||||
throw new NotImplementedException("TODO: session 22");
|
||||
|
||||
public void DeleteSession(MqttSession session) =>
|
||||
throw new NotImplementedException("TODO: session 22");
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// Global session manager stub
|
||||
// ============================================================================
|
||||
|
||||
/// <summary>
|
||||
/// Server-wide MQTT session manager.
|
||||
/// Mirrors Go <c>mqttSessionManager</c> struct in server/mqtt.go.
|
||||
/// </summary>
|
||||
internal sealed class MqttSessionManager
|
||||
{
|
||||
private readonly Lock _mu = new();
|
||||
|
||||
/// <summary>Per-account session managers keyed by account name.</summary>
|
||||
public Dictionary<string, MqttAccountSessionManager> Sessions { get; } = new();
|
||||
|
||||
/// <summary>Lock for this manager.</summary>
|
||||
public Lock Mu => _mu;
|
||||
}
|
||||
Reference in New Issue
Block a user