streams

package module

Go to main page

Versions in this module

v0

v0.2.6

May 4, 2022

v0.2.5

Apr 28, 2022

Changes in this version

type StreamRegistry

+ func NewStreamRegistry() StreamRegistry

v0.2.4

Apr 13, 2022

Changes in this version

type Hub

+ func (h *Hub) GetStreamReaderNodes(stream string) []ReaderNode

type ReaderNode

+ func GetStreamReaderNodes(stream string) []ReaderNode

v0.2.3

Apr 13, 2022

Changes in this version

type Message

+ StreamVersion int

type NewMessageArgs

+ StreamVersion int

type StreamMetadata

+ StreamVersion int

v0.2.2

Apr 12, 2022

v0.2.1

Apr 10, 2022

Changes in this version

+ const MarshalerAvroContentType

+ const MarshalerJSONContentType

+ const MarshalerProtoContentType

+ func Read(message interface{}, opts ...ReaderNodeOption) error

+ func ReadByStreamKey(stream string, opts ...ReaderNodeOption)

+ func RegisterStream(message interface{}, metadata StreamMetadata)

+ func RegisterStreamByString(messageType string, metadata StreamMetadata)

+ func Start(ctx context.Context)

+ func Write(ctx context.Context, message interface{}) error

+ func WriteBatch(ctx context.Context, messages ...interface{}) (uint32, error)

+ func WriteByMessageKey(ctx context.Context, messageKey string, message interface{}) error

+ func WriteByMessageKeyBatch(ctx context.Context, items WriteByMessageKeyBatchItems) (uint32, error)

+ func WriteRawMessage(ctx context.Context, message Message) error

+ func WriteRawMessageBatch(ctx context.Context, messages ...Message) (uint32, error)

type Hub

+ var DefaultHub *Hub

+ var ErrNilDefaultHub = errors.New("streams: DefaultHub has not been initialized")

v0.2.0

Apr 10, 2022

Changes in this version

+ const CloudEventsSpecVersion

+ var DefaultConcurrencyLevel = 1

+ var DefaultHubInstanceName = "com.streams"

+ var DefaultMaxHandlerPoolSize = 10

+ var DefaultRetryInitialInterval = time.Second * 3

+ var DefaultRetryMaxInterval = time.Second * 15

+ var DefaultRetryTimeout = time.Second * 15

+ var ErrInvalidProtocolBufferFormat = errors.New("streams: Invalid protocol buffer data")

+ var ErrMissingSchemaDefinition = errors.New("streams: Missing stream schema definition in schema registry")

+ var ErrMissingStream = errors.New("streams: Missing stream entry in stream registry")

+ var ErrMissingWriterDriver = errors.New("streams: Missing writer driver")

+ var ReaderBaseBehaviours = []ReaderBehaviour

+ var ReaderBaseBehavioursNoUnmarshal = []ReaderBehaviour

+ func InjectMessageCausationID(ctx context.Context, messageID string) string

+ func InjectMessageCorrelationID(ctx context.Context, messageID string) string

+ type AvroMarshaler struct

+ HashingFactory Hashing64AlgorithmFactory

+ func NewAvroMarshaler() AvroMarshaler

+ func (a AvroMarshaler) ContentType() string

+ func (a AvroMarshaler) Marshal(schemaDef string, data interface{}) (parsedData []byte, err error)

+ func (a AvroMarshaler) Unmarshal(schemaDef string, data []byte, ref interface{}) (err error)

+ type Event interface

+ GetSubject func() string

+ type FailingMarshalerNoop struct

+ func (f FailingMarshalerNoop) ContentType() string

+ func (f FailingMarshalerNoop) Marshal(_ string, _ interface{}) ([]byte, error)

+ func (f FailingMarshalerNoop) Unmarshal(_ string, _ []byte, _ interface{}) error

+ type Hashing64AlgorithmFactory func() hash.Hash64

+ var DefaultHashing64AlgorithmFactory Hashing64AlgorithmFactory = func() hash.Hash64 { ... }

+ type Hub struct

+ IDFactory IDFactoryFunc

+ InstanceName string

+ Marshaler Marshaler

+ Reader Reader

+ ReaderBaseOptions []ReaderNodeOption

+ ReaderBehaviours []ReaderBehaviour

+ SchemaRegistry SchemaRegistry

+ StreamRegistry StreamRegistry

+ Writer Writer

+ func NewHub(opts ...HubOption) *Hub

+ func (h *Hub) Read(message interface{}, opts ...ReaderNodeOption) error

+ func (h *Hub) ReadByStreamKey(stream string, opts ...ReaderNodeOption)

+ func (h *Hub) RegisterStream(message interface{}, metadata StreamMetadata)

+ func (h *Hub) RegisterStreamByString(messageType string, metadata StreamMetadata)

+ func (h *Hub) Start(ctx context.Context)

+ func (h *Hub) Write(ctx context.Context, message interface{}) error

+ func (h *Hub) WriteBatch(ctx context.Context, messages ...interface{}) (uint32, error)

+ func (h *Hub) WriteByMessageKey(ctx context.Context, messageKey string, message interface{}) error

+ func (h *Hub) WriteByMessageKeyBatch(ctx context.Context, items WriteByMessageKeyBatchItems) (uint32, error)

+ func (h *Hub) WriteRawMessage(ctx context.Context, message Message) error

+ func (h *Hub) WriteRawMessageBatch(ctx context.Context, messages ...Message) (uint32, error)

+ type HubOption interface

+ func WithIDFactory(f IDFactoryFunc) HubOption

+ func WithInstanceName(n string) HubOption

+ func WithMarshaler(m Marshaler) HubOption

+ func WithReader(d Reader) HubOption

+ func WithReaderBaseOptions(opts ...ReaderNodeOption) HubOption

+ func WithReaderBehaviours(b ...ReaderBehaviour) HubOption

+ func WithSchemaRegistry(r SchemaRegistry) HubOption

+ func WithWriter(p Writer) HubOption

+ type IDFactoryFunc func() (string, error)

+ var RandInt64Factory IDFactoryFunc = func() (string, error) { ... }

+ var UuidIdFactory IDFactoryFunc = func() (string, error) { ... }

+ type InMemorySchemaRegistry map[string]string

+ func (i InMemorySchemaRegistry) GetSchemaDefinition(name string, version int) (string, error)

+ func (i InMemorySchemaRegistry) RegisterDefinition(name, def string, version int)

+ type JSONMarshaler struct

+ func (m JSONMarshaler) ContentType() string

+ func (m JSONMarshaler) Marshal(_ string, data interface{}) ([]byte, error)

+ func (m JSONMarshaler) Unmarshal(_ string, data []byte, ref interface{}) error

+ type Marshaler interface

+ ContentType func() string

+ Marshal func(schemaDef string, data interface{}) ([]byte, error)

+ Unmarshal func(schemaDef string, data []byte, ref interface{}) error

+ type Message struct

+ CausationID string

+ CorrelationID string

+ Data []byte

+ DataContentType string

+ DataSchema string

+ DataSchemaVersion int

+ DecodedData interface{}

+ GroupName string

+ ID string

+ Source string

+ SpecVersion string

+ Stream string

+ Subject string

+ Timestamp string

+ Type string

+ func NewMessage(args NewMessageArgs) Message

+ type MessageContextKey string

+ const ContextCausationID

+ const ContextCorrelationID

+ type NewMessageArgs struct

+ ContentType string

+ Data []byte

+ GroupName string

+ ID string

+ SchemaDefinitionName string

+ SchemaVersion int

+ Source string

+ Stream string

+ Subject string

+ type NoopSchemaRegistry struct

+ func (n NoopSchemaRegistry) GetSchemaDefinition(_ string, _ int) (string, error)

+ type ProtocolBuffersMarshaler struct

+ func (p ProtocolBuffersMarshaler) ContentType() string

+ func (p ProtocolBuffersMarshaler) Marshal(_ string, data interface{}) ([]byte, error)

+ func (p ProtocolBuffersMarshaler) Unmarshal(_ string, data []byte, ref interface{}) error

+ type Reader interface

+ ExecuteTask func(_ context.Context, _ ReaderTask) error

+ type ReaderBehaviour func(node *ReaderNode, hub *Hub, next ReaderHandleFunc) ReaderHandleFunc

+ type ReaderHandleFunc func(context.Context, Message) error

+ type ReaderHandler interface

+ Read func(context.Context, Message) error

+ type ReaderHandlerNoop struct

+ func (l ReaderHandlerNoop) Read(_ context.Context, _ Message) error

+ type ReaderNode struct

+ ConcurrencyLevel int

+ Group string

+ HandlerFunc ReaderHandleFunc

+ MaxHandlerPoolSize int

+ ProviderConfiguration interface{}

+ Reader Reader

+ RetryInitialInterval time.Duration

+ RetryMaxInterval time.Duration

+ RetryTimeout time.Duration

+ Stream string

+ type ReaderNodeOption interface

+ func WithConcurrencyLevel(n int) ReaderNodeOption

+ func WithDriver(d Reader) ReaderNodeOption

+ func WithGroup(g string) ReaderNodeOption

+ func WithHandler(l ReaderHandler) ReaderNodeOption

+ func WithHandlerFunc(l ReaderHandleFunc) ReaderNodeOption

+ func WithMaxHandlerPoolSize(n int) ReaderNodeOption

+ func WithProviderConfiguration(cfg interface{}) ReaderNodeOption

+ func WithRetryInitialInterval(d time.Duration) ReaderNodeOption

+ func WithRetryMaxInterval(d time.Duration) ReaderNodeOption

+ func WithRetryTimeout(d time.Duration) ReaderNodeOption

+ type ReaderTask struct

+ Configuration interface{}

+ Group string

+ HandlerFunc ReaderHandleFunc

+ MaxHandlerPoolSize int

+ Stream string

+ Timeout time.Duration

+ type SchemaRegistry interface

+ GetSchemaDefinition func(name string, version int) (string, error)

+ type StreamMetadata struct

+ GoType reflect2.Type

+ SchemaDefinitionName string

+ SchemaVersion int

+ Stream string

+ type StreamRegistry map[string]StreamMetadata

+ func (r StreamRegistry) Get(message interface{}) (StreamMetadata, error)

+ func (r StreamRegistry) GetByStreamName(name string) (StreamMetadata, error)

+ func (r StreamRegistry) GetByString(key string) (StreamMetadata, error)

+ func (r StreamRegistry) Set(message interface{}, metadata StreamMetadata)

+ func (r StreamRegistry) SetByString(key string, metadata StreamMetadata)

+ type WriteByMessageKeyBatchItems map[string]interface

+ type Writer interface

+ Write func(ctx context.Context, message Message) error

+ WriteBatch func(ctx context.Context, messages ...Message) (uint32, error)

+ var NoopWriter Writer = noopWriter{}

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL