70 lines
2.1 KiB
Go
70 lines
2.1 KiB
Go
package process
|
|
|
|
import (
|
|
"context"
|
|
"crypto/rand"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go"
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
|
|
stdio "git.novox.be/novox/mesh-sdk/go"
|
|
)
|
|
|
|
// Emit publishes an event as this process's module (novox/hq ADR 0436 clause 4), under its module's own
|
|
// namespace (mesh.mod.<module>.event.<type>, the same subject the module's own principal would publish it
|
|
// on), with the SDK's own envelope and headers (stdio.go:498-509). A chosen message id
|
|
// (EmitOptions.MessageID) is sent as x-event-id, and as the bus's own dedup id (Nats-Msg-Id), so a second
|
|
// publish inside the stream's duplicate window is dropped; empty is a random one, as stdio.Emit on main.
|
|
func (c *Conn) Emit(eventType string, body any, opts ...stdio.EmitOptions) error {
|
|
data, err := json.Marshal(body)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
id := ""
|
|
for _, o := range opts {
|
|
if o.MessageID != "" {
|
|
id = o.MessageID
|
|
}
|
|
}
|
|
if id == "" {
|
|
id = eventID()
|
|
}
|
|
msg := nats.NewMsg("mesh.mod." + c.account.Module + ".event." + eventType)
|
|
msg.Header.Set("x-event-id", id)
|
|
msg.Header.Set("x-source", c.account.Module)
|
|
msg.Header.Set("x-node", c.account.Node)
|
|
msg.Header.Set("x-time", time.Now().UTC().Format(time.RFC3339Nano))
|
|
msg.Header.Set("content-type", "application/json")
|
|
for _, o := range opts {
|
|
if o.CausationID != "" {
|
|
msg.Header.Set("x-causation-id", o.CausationID)
|
|
}
|
|
if o.Schema != "" {
|
|
msg.Header.Set("x-schema", o.Schema)
|
|
}
|
|
}
|
|
msg.Data = data
|
|
js, err := jetstream.New(c.nc)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
ctx, cancel := context.WithTimeout(context.Background(), stateTimeout)
|
|
defer cancel()
|
|
_, err = js.PublishMsg(ctx, msg, jetstream.WithMsgID(id))
|
|
return err
|
|
}
|
|
|
|
// eventID is a random UUID, version 4, as stdio's own eventID (go/stdio.go:515-522) — duplicated rather
|
|
// than imported, since stdio's is unexported and this package touches none of stdio's bundle machinery.
|
|
func eventID() string {
|
|
var b [16]byte
|
|
_, _ = rand.Read(b[:])
|
|
b[6] = (b[6] & 0x0f) | 0x40
|
|
b[8] = (b[8] & 0x3f) | 0x80
|
|
h := hex.EncodeToString(b[:])
|
|
return h[0:8] + "-" + h[8:12] + "-" + h[12:16] + "-" + h[16:20] + "-" + h[20:32]
|
|
}
|