messaging

package
v0.0.179 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Nov 11, 2026 License: Apache-2.0 Imports: 18 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	MsgPropTimestamp = "timestamp"
	// MsgPropCluster is the property name for the cluster name
	MsgPropCluster = "cluster"
	// MsgPropAccount is the property name for the account name
	MsgPropAccount = "account"
	// MsgPropEvent is the property name for the event type
	MsgPropEvent = "event"
	// MsgPropResourceKindGroup is the property name for the API Group of the resource
	MsgPropResourceKindGroup = "group"
	// MsgPropResourceKindVersion is the property name for the API Version of the resource
	MsgPropResourceKindVersion = "version"
	// MsgPropResourceKindResource is the property name for the API Resource of the resource
	MsgPropResourceKindResource = "resource"
	// MsgPropResourceName is the property name for the name of the resource
	MsgPropResourceName = "name"
	// MsgPropResourceNamespace is the property name for the namespace of the resource
	MsgPropResourceNamespace = "namespace"
	// MsgPropResourceVersion is the property name for the resource version of the resource
	MsgPropResourceVersion = "resourceVersion"
	// MsgPropProducerSource identifies the message producer (Pulsar property / Kafka header)
	MsgPropProducerSource = "x-producer-source"
	// MsgPropProducerSourceSynchronizerServer is the producer source value for synchronizer server outbound messages
	MsgPropProducerSourceSynchronizerServer = "synchronizer-server"
	// SynchronizerServerProducerKey is the legacy Pulsar message key used for self-loop filtering
	SynchronizerServerProducerKey = "SynchronizerServerProducer"

	MsgPropEventValueGetObjectMessage                  = "GetObject"
	MsgPropEventValuePatchObjectMessage                = "PatchObject"
	MsgPropEventValueVerifyObjectMessage               = "VerifyObject"
	MsgPropEventValueDeleteObjectMessage               = "DeleteObject"
	MsgPropEventValuePutObjectMessage                  = "PutObject"
	MsgPropEventValueServerConnectedMessage            = "ServerConnected"
	MsgPropEventValueReconciliationRequestMessage      = "ReconciliationRequest"
	MsgPropEventValueReconciliationRequestMessageBatch = "ReconciliationRequestBatch"
	MsgPropEventValueConnectedClientsMessage           = "ConnectedClients"
)

Variables

This section is empty.

Functions

func BuildProducerProperties added in v0.0.155 

func BuildProducerProperties(account, cluster, eventType string, optionalProperties ...map[string]string) map[string]string

BuildProducerProperties builds the standard property map for outbound synchronizer server messages.

func IsSelfProducedMessage added in v0.0.155 

func IsSelfProducedMessage(properties map[string]string, messageKey string) bool

IsSelfProducedMessage returns true if the message was produced by the synchronizer server.

func KindNameToCustomProperties added in v0.0.95 

func KindNameToCustomProperties(kind domain.KindName) map[string]string

func ReaderWorkers added in v0.0.179 

func ReaderWorkers(adapter adapters.Adapter, configured int) int

ReaderWorkers keeps broker receive order when the adapter needs ordered replay. Otherwise the configured worker count retains its existing meaning.

func RecordProducerMessage added in v0.0.155 

func RecordProducerMessage(backend, status string, properties map[string]string, payloadBytes int)

RecordProducerMessage increments backend-agnostic and Pulsar-alias producer metrics.

func RegisterFromConfigFactory added in v0.0.155 

func RegisterFromConfigFactory(f factoryFunc)

RegisterFromConfigFactory registers the config-driven factory implementation. Called from backend init to avoid an import cycle.

Types

type BackendMessageRouter added in v0.0.179 

type BackendMessageRouter interface {
	HandleBackendMessage(context.Context, adapters.Adapter, IncomingMessage, *MessageHandler) error
}

BackendMessageRouter serializes message routing with connection publication. Adapters without replay retain the ordinary delivery path.

type Components added in v0.0.155 

type Components struct {
	Producer MessageProducer
	Reader   MessageReader
	Close    func()
}

func NewFromConfig added in v0.0.155 

func NewFromConfig(ctx context.Context, cfg config.Config) (*Components, error)

NewFromConfig creates message queue components from configuration. The context bounds any connection attempt made while building them, so a shutdown signal during startup is not ignored. Returns (nil, nil) when no message queue backend is configured.

func (*Components) Shutdown added in v0.0.160 

func (c *Components) Shutdown()

Shutdown runs the backend cleanup, which flushes buffered producer records instead of dropping them. Does nothing on a nil *Components (no queue configured) or a nil Close.

type ConnectedClient added in v0.0.64 

type ConnectedClient struct {
	Cluster             string    `json:"cluster"`
	Account             string    `json:"account"`
	SynchronizerVersion string    `json:"synchronizerVersion"`
	HelmVersion         string    `json:"helmVersion"`
	ConnectionId        string    `json:"connectionId"`
	ConnectionTime      time.Time `json:"connectionTime"`
	GitVersion          string    `json:"gitVersion"`
	CloudProvider       string    `json:"cloudProvider"`
	ClusterUID          string    `json:"clusterUID"`
	ResourceGroup       string    `json:"resourceGroup,omitempty"`
}

type ConnectedClientsMessage added in v0.0.64 

type ConnectedClientsMessage struct {
	ServerName string            `json:"serverName"`
	Clients    []ConnectedClient `json:"clients"`
	Timestamp  time.Time         `json:"keepalive"`
	Depth      int               `json:"depth"`
	MsgId      string            `json:"msgId"`
}

type DeleteObjectMessage 

type DeleteObjectMessage struct {
	Cluster         string `json:"cluster"`
	Account         string `json:"account"`
	Depth           int    `json:"depth"`
	Kind            string `json:"kind"`
	MsgId           string `json:"msgId"`
	Name            string `json:"name"`
	Namespace       string `json:"namespace"`
	ResourceVersion int    `json:"resourceVersion"`
}

type GetObjectMessage 

type GetObjectMessage struct {
	BaseObject      []byte `json:"baseObject"`
	Cluster         string `json:"cluster"`
	Account         string `json:"account"`
	Depth           int    `json:"depth"`
	Kind            string `json:"kind"`
	MsgId           string `json:"msgId"`
	Name            string `json:"name"`
	Namespace       string `json:"namespace"`
	ResourceVersion int    `json:"resourceVersion"`
}

type IncomingMessage added in v0.0.155 

type IncomingMessage struct {
	ID         string
	Properties map[string]string
	Payload    []byte
}

IncomingMessage is a backend-neutral representation of a message consumed by the synchronizer server.

type MessageHandler added in v0.0.155 

type MessageHandler struct {
	Name string
	// contains filtered or unexported fields
}

func (*MessageHandler) Deliver added in v0.0.179 

func (h *MessageHandler) Deliver(ctx context.Context, adapter adapters.Adapter, msg IncomingMessage) (result error)

Deliver dispatches a message without entering the routing hook again. A replay router calls it only while it owns the client's connection routing lock.

func (*MessageHandler) Handle added in v0.0.155 

func (h *MessageHandler) Handle(ctx context.Context, adapter adapters.Adapter, msg IncomingMessage) error

type MessageProducer 

type MessageProducer interface {
	// ProduceMessage produces a message to a messaging system
	ProduceMessage(ctx context.Context, id domain.ClientIdentifier, eventType string, payload []byte, optionalProperties ...map[string]string) error
	ProduceMessageWithoutIdentifier(ctx context.Context, eventType string, payload []byte) error
}

type MessageReader added in v0.0.40 

type MessageReader interface {
	// Start starts the message reader and blocks until the context is done
	Start(mainContext context.Context, adapter adapters.Adapter)
}

type NewChecksumMessage 

type NewChecksumMessage struct {
	Checksum        string `json:"checksum"`
	Cluster         string `json:"cluster"`
	Account         string `json:"account"`
	Depth           int    `json:"depth"`
	Kind            string `json:"kind"`
	MsgId           string `json:"msgId"`
	Name            string `json:"name"`
	Namespace       string `json:"namespace"`
	ResourceVersion int    `json:"resourceVersion"`
}

type NewObjectMessage 

type NewObjectMessage struct {
	Cluster         string `json:"cluster"`
	Account         string `json:"account"`
	Depth           int    `json:"depth"`
	Kind            string `json:"kind"`
	MsgId           string `json:"msgId"`
	Name            string `json:"name"`
	Namespace       string `json:"namespace"`
	Object          []byte `json:"patch"`
	ResourceVersion int    `json:"resourceVersion"`
}

type PatchObjectMessage 

type PatchObjectMessage struct {
	Checksum        string `json:"checksum"`
	Cluster         string `json:"cluster"`
	Account         string `json:"account"`
	Depth           int    `json:"depth"`
	Kind            string `json:"kind"`
	MsgId           string `json:"msgId"`
	Name            string `json:"name"`
	Namespace       string `json:"namespace"`
	Patch           []byte `json:"patch"`
	ResourceVersion int    `json:"resourceVersion"`
}

type PutObjectMessage 

type PutObjectMessage struct {
	Checksum        string `json:"checksum"`
	Cluster         string `json:"cluster"`
	Account         string `json:"account"`
	Depth           int    `json:"depth"`
	Kind            string `json:"kind"`
	MsgId           string `json:"msgId"`
	Name            string `json:"name"`
	Namespace       string `json:"namespace"`
	Object          []byte `json:"patch"`
	ResourceVersion int    `json:"resourceVersion"`
}

type ReconciliationRequestMessage added in v0.0.47 

type ReconciliationRequestMessage struct {
	Cluster         string                                   `json:"cluster"`
	Account         string                                   `json:"account"`
	Depth           int                                      `json:"depth"`
	MsgId           string                                   `json:"msgId"`
	ServerInitiated bool                                     `json:"serverInitiated"`
	KindToObjects   map[string][]ReconciliationRequestObject `json:"kindToObject"`
}

type ReconciliationRequestMessageBatch added in v0.0.110 

type ReconciliationRequestMessageBatch struct {
	Depth    int                            `json:"depth"`
	MsgId    string                         `json:"msgId"`
	Requests []ReconciliationRequestMessage `json:"requests"`
}

type ReconciliationRequestObject added in v0.0.47 

type ReconciliationRequestObject struct {
	Checksum        string `json:"checksum"`
	ResourceVersion int    `json:"resourceVersion"`
	Name            string `json:"name"`
	Namespace       string `json:"namespace"`
}

type ServerConnectedMessage 

type ServerConnectedMessage struct {
	Cluster string `json:"cluster"`
	Account string `json:"account"`
	Depth   int    `json:"depth"`
	MsgId   string `json:"msgId"`
}

type VerifyObjectMessage 

type VerifyObjectMessage struct {
	Checksum        string `json:"checksum"`
	Cluster         string `json:"cluster"`
	Account         string `json:"account"`
	Depth           int    `json:"depth"`
	Kind            string `json:"kind"`
	MsgId           string `json:"msgId"`
	Name            string `json:"name"`
	Namespace       string `json:"namespace"`
	ResourceVersion int    `json:"resourceVersion"`
}

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.