- AtomicBatchPublishEngine: stage/commit/rollback semantics for batch publish - JsVersioning: API level negotiation and stream/consumer metadata - Fix NormalizeConfig missing AllowAtomicPublish, Metadata, PersistMode copy - 46 batch publish tests + 67 versioning tests, all passing
107 lines
4.5 KiB
C#
107 lines
4.5 KiB
C#
namespace NATS.Server.JetStream.Models;
|
|
|
|
public sealed class StreamConfig
|
|
{
|
|
public string Name { get; set; } = string.Empty;
|
|
public string Description { get; set; } = string.Empty;
|
|
public List<string> Subjects { get; set; } = [];
|
|
public int MaxMsgs { get; set; }
|
|
public long MaxBytes { get; set; }
|
|
public int MaxMsgsPer { get; set; }
|
|
public int MaxAgeMs { get; set; }
|
|
public int MaxMsgSize { get; set; }
|
|
public int MaxConsumers { get; set; }
|
|
public int DuplicateWindowMs { get; set; }
|
|
public bool Sealed { get; set; }
|
|
public bool DenyDelete { get; set; }
|
|
public bool DenyPurge { get; set; }
|
|
public bool AllowDirect { get; set; }
|
|
// Go: StreamConfig.AllowMsgTTL — per-message TTL header support
|
|
public bool AllowMsgTtl { get; set; }
|
|
// Go: StreamConfig.FirstSeq — initial sequence number for the stream
|
|
public ulong FirstSeq { get; set; }
|
|
public RetentionPolicy Retention { get; set; } = RetentionPolicy.Limits;
|
|
public DiscardPolicy Discard { get; set; } = DiscardPolicy.Old;
|
|
public StorageType Storage { get; set; } = StorageType.Memory;
|
|
public int Replicas { get; set; } = 1;
|
|
public string? Mirror { get; set; }
|
|
public string? Source { get; set; }
|
|
public List<StreamSourceConfig> Sources { get; set; } = [];
|
|
|
|
// Go: StreamConfig.SubjectTransform — transforms inbound message subjects on store.
|
|
// Source and Dest follow the same token-wildcard rules as NATS subject transforms.
|
|
// Go reference: server/stream.go:352 (SubjectTransform field in StreamConfig)
|
|
public string? SubjectTransformSource { get; set; }
|
|
public string? SubjectTransformDest { get; set; }
|
|
|
|
// Go: StreamConfig.RePublish — re-publish stored messages on a separate subject.
|
|
// Source is the filter (empty = match all); Dest is the target subject pattern.
|
|
// Go reference: server/stream.go:356 (RePublish field in StreamConfig)
|
|
public string? RePublishSource { get; set; }
|
|
public string? RePublishDest { get; set; }
|
|
// Go: RePublish.HeadersOnly — republished copy omits message body.
|
|
public bool RePublishHeadersOnly { get; set; }
|
|
|
|
// Go: StreamConfig.SubjectDeleteMarkerTTL — duration to retain delete markers.
|
|
// When > 0 and AllowMsgTTL is true, expired messages emit a delete-marker msg.
|
|
// Incompatible with Mirror config.
|
|
// Go reference: server/stream.go:361 (SubjectDeleteMarkerTTL field)
|
|
public int SubjectDeleteMarkerTtlMs { get; set; }
|
|
|
|
// Go: StreamConfig.AllowMsgSchedules — enables scheduled publish headers.
|
|
// Incompatible with Mirror and Sources.
|
|
// Go reference: server/stream.go:369 (AllowMsgSchedules field)
|
|
public bool AllowMsgSchedules { get; set; }
|
|
|
|
// Go: StreamConfig.AllowMsgCounter — enables CRDT counter semantics on messages.
|
|
// Added in v2.12, requires API level 2.
|
|
// Go reference: server/stream.go:365 (AllowMsgCounter field)
|
|
public bool AllowMsgCounter { get; set; }
|
|
|
|
// Go: StreamConfig.AllowAtomicPublish — enables atomic batch publishing.
|
|
// Added in v2.12, requires API level 2.
|
|
// Go reference: server/stream.go:367 (AllowAtomicPublish field)
|
|
public bool AllowAtomicPublish { get; set; }
|
|
|
|
// Go: StreamConfig.PersistMode — async vs sync storage persistence.
|
|
// AsyncPersistMode requires API level 2.
|
|
// Go reference: server/stream.go:375 (PersistMode field)
|
|
public PersistMode PersistMode { get; set; } = PersistMode.Sync;
|
|
|
|
// Go: StreamConfig.Metadata — user-supplied and server-managed key/value metadata.
|
|
// The server automatically sets _nats.req.level, _nats.ver, _nats.level.
|
|
// Go reference: server/stream.go:380 (Metadata field)
|
|
public Dictionary<string, string>? Metadata { get; set; }
|
|
}
|
|
|
|
/// <summary>
|
|
/// Persistence mode for the stream.
|
|
/// Go reference: server/stream.go — AsyncPersistMode constant.
|
|
/// </summary>
|
|
public enum PersistMode
|
|
{
|
|
Sync = 0,
|
|
Async = 1,
|
|
}
|
|
|
|
public enum StorageType
|
|
{
|
|
Memory,
|
|
File,
|
|
}
|
|
|
|
public sealed class StreamSourceConfig
|
|
{
|
|
public string Name { get; set; } = string.Empty;
|
|
public string? SubjectTransformPrefix { get; set; }
|
|
public string? SourceAccount { get; set; }
|
|
|
|
// Go: StreamSource.FilterSubject — only forward messages matching this subject filter.
|
|
public string? FilterSubject { get; set; }
|
|
|
|
// Deduplication window in milliseconds for Nats-Msg-Id header-based dedup.
|
|
// Defaults to 0 (disabled). When > 0, duplicate messages with the same Nats-Msg-Id
|
|
// within this window are silently dropped.
|
|
public int DuplicateWindowMs { get; set; }
|
|
}
|