116 lines
3.9 KiB
Go
116 lines
3.9 KiB
Go
package buscompose
|
|
|
|
import (
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
)
|
|
|
|
// The streams a manifest declares, as the server is asked to create them (novox/hq ADR 0547 §5, issue 1790): the
|
|
// grants part creates each one the bus does not hold yet, with what the controller asserts at its start, so the two
|
|
// make the same stream. Pure. Only a create is ever asked with these: a stream the bus holds is never updated by the
|
|
// grants part (CreateStreamGrants grants no update).
|
|
|
|
// BucketStreamConfig is a module's bucket as the stream the controller's CreateOrUpdateKeyValue makes of its options
|
|
// (mesh-controller internal/broker BucketConfig, with no marker age held): the client library's key-value shape
|
|
// (rollup, no delete, direct reads, discard new, a duplicate window of two minutes or the age when shorter) with the
|
|
// mesh's caps, file storage and one replica.
|
|
func BucketStreamConfig(b Bucket) jetstream.StreamConfig {
|
|
history := b.History
|
|
if history < 1 {
|
|
history = 1
|
|
}
|
|
age := time.Duration(b.TTLSeconds) * time.Second
|
|
dupes := 2 * time.Minute
|
|
if age > 0 && age < dupes {
|
|
dupes = age
|
|
}
|
|
cfg := jetstream.StreamConfig{
|
|
Name: "KV_" + b.Bucket(),
|
|
Description: b.Why(),
|
|
Subjects: []string{"$KV." + b.Bucket() + ".>"},
|
|
MaxMsgsPerSubject: int64(history),
|
|
MaxBytes: StateMaxBytes,
|
|
MaxAge: age,
|
|
MaxMsgSize: StateMaxValueBytes,
|
|
Storage: jetstream.FileStorage,
|
|
Replicas: 1,
|
|
AllowRollup: true,
|
|
DenyDelete: true,
|
|
Duplicates: dupes,
|
|
MaxMsgs: -1,
|
|
MaxConsumers: -1,
|
|
AllowDirect: true,
|
|
Discard: jetstream.DiscardNew,
|
|
}
|
|
if b.KeyTTL {
|
|
cfg.AllowMsgTTL, cfg.SubjectDeleteMarkerTTL = true, KeyTTLMarkerTTL
|
|
}
|
|
return cfg
|
|
}
|
|
|
|
// ModuleStreamConfig is a module's own stream as the controller asserts it (mesh-controller internal/broker
|
|
// StreamConfigFor of ModuleStream.Stream).
|
|
func ModuleStreamConfig(m ModuleStream) jetstream.StreamConfig {
|
|
s := m.Stream()
|
|
cfg := jetstream.StreamConfig{
|
|
Name: s.Name,
|
|
Subjects: s.Subjects,
|
|
Retention: jetstream.LimitsPolicy,
|
|
MaxAge: time.Duration(s.MaxAge) * time.Second,
|
|
MaxMsgsPerSubject: int64(s.MaxMsgsPerSubject),
|
|
Description: s.Why,
|
|
AllowDirect: s.Direct,
|
|
AllowMsgTTL: s.AllowMsgTTL,
|
|
}
|
|
if s.Retention == RetentionWorkQueue {
|
|
cfg.Retention = jetstream.WorkQueuePolicy
|
|
}
|
|
if s.Restricted || s.WorkQueue {
|
|
cfg.Metadata = map[string]string{}
|
|
if s.Restricted {
|
|
cfg.Metadata[RestrictedMetadata] = "true"
|
|
}
|
|
if s.WorkQueue {
|
|
cfg.Metadata[WorkQueueMetadata] = "true"
|
|
}
|
|
}
|
|
if s.MaxBytes > 0 {
|
|
cfg.MaxBytes = s.MaxBytes
|
|
}
|
|
if s.DiscardNew {
|
|
cfg.Discard = jetstream.DiscardNew
|
|
}
|
|
if s.DuplicatesSeconds > 0 {
|
|
cfg.Duplicates = time.Duration(s.DuplicatesSeconds) * time.Second
|
|
}
|
|
return cfg
|
|
}
|
|
|
|
// ConsumerConfig is a pull consumer as the controller's EnsureConsumer creates it (mesh-controller internal/broker
|
|
// JetStream.EnsureConsumer, a consumer not there yet): its durable name, explicit acks, its ack wait, max deliver and max
|
|
// ack pending, its description, one filter or several, and from the stream's end when it is made FromNow. The grants part
|
|
// creates a module's and a process's consumers with it from the lease write (novox/hq ADR 0561 §6): a create alone,
|
|
// never an update of one the bus holds. A push consumer or a queue group is the controller's own and never made here.
|
|
func ConsumerConfig(c Consumer) jetstream.ConsumerConfig {
|
|
cfg := jetstream.ConsumerConfig{
|
|
Durable: c.Name,
|
|
AckPolicy: jetstream.AckExplicitPolicy,
|
|
AckWait: time.Duration(c.AckWaitSeconds) * time.Second,
|
|
MaxDeliver: c.MaxDeliver,
|
|
MaxAckPending: c.MaxAckPending,
|
|
Description: c.Why,
|
|
}
|
|
switch len(c.Filters) {
|
|
case 0:
|
|
case 1:
|
|
cfg.FilterSubject = c.Filters[0]
|
|
default:
|
|
cfg.FilterSubjects = c.Filters
|
|
}
|
|
if c.FromNow {
|
|
cfg.DeliverPolicy = jetstream.DeliverNewPolicy
|
|
}
|
|
return cfg
|
|
}
|