Why Go
- Case Studies
  
  Common problems companies solve with Go
- Use Cases
  
  Stories about how and why companies use Go
- Security Policy
  
  How Go can help keep you secure by default
Learn
Docs
- Effective Go
  
  Tips for writing clear, performant, and idiomatic Go code
- Go User Manual
  
  A complete introduction to building software with Go
- Standard library
  
  Reference documentation for Go's standard library
- Release Notes
  
  Learn what's new in each Go release
Packages
Community
- Recorded Talks
  
  Videos from prior events
- Meetups
  
  Meet other local Go developers
- Conferences
  
  Learn and network with Go developers from around the world
- Go blog
  
  The Go project's official blog.
- Go project
  
  Get help and stay informed from Go
- Get connected

streamhub

package module

Go to main page

Versions in this module

v0

v0.1.5

Mar 28, 2022

Changes in this version

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

type Hub

+ Writer Writer

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

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

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

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

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

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

type HubOption

+ func WithWriter(p Writer) HubOption

+ type WriteByMessageKeyBatchItems map[string]interface

+ type Writer interface

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

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

+ var NoopWriter Writer = noopWriter{}

v0.1.4

Mar 13, 2022

v0.1.3

Mar 13, 2022

v0.1.2

Mar 10, 2022

v0.1.1

Mar 10, 2022

v0.1.0

Mar 5, 2022

Changes in this version

+ const CloudEventsSpecVersion

+ var DefaultConcurrencyLevel = 1

+ var DefaultHubInstanceName = "com.streamhub"

+ var DefaultMaxHandlerPoolSize = 10

+ var DefaultRetryInitialInterval = time.Second * 3

+ var DefaultRetryMaxInterval = time.Second * 15

+ var DefaultRetryTimeout = time.Second * 15

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

+ var ErrMissingPublisherDriver = errors.New("streamhub: Missing publisher driver")

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

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

+ var ListenerBaseBehaviours = []ListenerBehaviour

+ var ListenerBaseBehavioursNoUnmarshal = []ListenerBehaviour

+ 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

+ ListenerBaseOptions []ListenerNodeOption

+ ListenerBehaviours []ListenerBehaviour

+ ListenerDriver ListenerDriver

+ Marshaler Marshaler

+ Publisher Publisher

+ SchemaRegistry SchemaRegistry

+ StreamRegistry StreamRegistry

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

+ func (h *Hub) Listen(message interface{}, opts ...ListenerNodeOption) error

+ func (h *Hub) ListenByStreamKey(stream string, opts ...ListenerNodeOption)

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

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

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

+ func (h *Hub) PublishByMessageKeyBatch(ctx context.Context, items PublishByMessageKeyBatchItems) error

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

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

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

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

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

+ type HubOption interface

+ func WithIDFactory(f IDFactoryFunc) HubOption

+ func WithInstanceName(n string) HubOption

+ func WithListenerBaseOptions(opts ...ListenerNodeOption) HubOption

+ func WithListenerBehaviours(b ...ListenerBehaviour) HubOption

+ func WithListenerDriver(d ListenerDriver) HubOption

+ func WithMarshaler(m Marshaler) HubOption

+ func WithPublisher(p Publisher) HubOption

+ func WithSchemaRegistry(r SchemaRegistry) 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 Listener interface

+ Listen func(context.Context, Message) error

+ type ListenerBehaviour func(node *ListenerNode, hub *Hub, next ListenerFunc) ListenerFunc

+ type ListenerDriver interface

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

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

+ type ListenerNode struct

+ ConcurrencyLevel int

+ Group string

+ HandlerFunc ListenerFunc

+ ListenerDriver ListenerDriver

+ MaxHandlerPoolSize int

+ ProviderConfiguration interface{}

+ RetryInitialInterval time.Duration

+ RetryMaxInterval time.Duration

+ RetryTimeout time.Duration

+ Stream string

+ type ListenerNodeOption interface

+ func WithConcurrencyLevel(n int) ListenerNodeOption

+ func WithDriver(d ListenerDriver) ListenerNodeOption

+ func WithGroup(g string) ListenerNodeOption

+ func WithListener(l Listener) ListenerNodeOption

+ func WithListenerFunc(l ListenerFunc) ListenerNodeOption

+ func WithMaxHandlerPoolSize(n int) ListenerNodeOption

+ func WithProviderConfiguration(cfg interface{}) ListenerNodeOption

+ func WithRetryInitialInterval(d time.Duration) ListenerNodeOption

+ func WithRetryMaxInterval(d time.Duration) ListenerNodeOption

+ func WithRetryTimeout(d time.Duration) ListenerNodeOption

+ type ListenerNoop struct

+ func (l ListenerNoop) Listen(_ context.Context, _ Message) error

+ type ListenerTask struct

+ Configuration interface{}

+ Group string

+ HandlerFunc ListenerFunc

+ MaxHandlerPoolSize int

+ Stream string

+ Timeout time.Duration

+ 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 PublishByMessageKeyBatchItems map[string]interface

+ type Publisher interface

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

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

+ var NoopPublisher Publisher = noopPublisher{}

+ 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)