subscription

Documentation

Index

• Constants
• Variables
• type Option
  • func WithAutoACK() Option
  • func WithCount() Option
  • func WithMiddlewares(handlers ...func(context.Context, *protocol.CloudEventsEnvelope)) Option
  • func WithWebHook(webhookHandler func(context.Context, *protocol.CloudEventsEnvelope), ssl bool) Option
• type ReceiveSettings
• type Subscription
  • func NewSubscription(topicName string, driverMetadata protocol.Metadata, options ...Option) (*Subscription, error)
  • func (s *Subscription) Receive(ctx context.Context, r *protocol.SubscribeRequest, ...) error

Constants

const (
	DefaultWebHookRequestTimeout = 60 * time.Second
)

Variables

var DefaultRecieveSettings = ReceiveSettings{

	RetryParams: &retry.DefaultRetryParams,

	MaxOutstandingMessages: 1000,

	WebHookRequestTimeout: DefaultWebHookRequestTimeout,
}

DefaultPublishSettings holds the default values for topics' Settings.

Types

type Option

type Option func(*Subscription) error

func WithAutoACK

func WithAutoACK() Option

func WithCount

func WithCount() Option

func WithMiddlewares

func WithMiddlewares(handlers ...func(context.Context, *protocol.CloudEventsEnvelope)) Option

func WithWebHook

func WithWebHook(webhookHandler func(context.Context, *protocol.CloudEventsEnvelope), ssl bool) Option

WithWebHook would turn on the ack function.

type ReceiveSettings

type ReceiveSettings struct {
	// EnableMessageOrdering enables protocol ordering.
	//
	// It is EXPERIMENTAL and a part of a closed alpha that may not be
	// accessible to all users. This field is subject to change or removal
	// without notice.
	EnableMessageOrdering bool

	EnableAck bool

	// RetryPolicy specifies how Cloud Pub/Sub retries protocol delivery.
	RetryParams *retry.Params

	// MaxOutstandingMessages is the maximum number of unprocessed messages
	// (unacknowledged but not yet expired). If MaxOutstandingMessages is 0, it
	// will be treated as if it were DefaultReceiveSettings.MaxOutstandingMessages.
	// If the value is negative, then there will be no limit on the number of
	// unprocessed messages.
	MaxOutstandingMessages int

	// WebHookRequestTimeout is the timeout when Subscription calls protocol's callback webhook via fasthttp.Client.
	WebHookRequestTimeout time.Duration
}

could add toproto() protoToSubscriptionConfig() from https://github.com/googleapis/google-cloud-go/blob/master/pubsub/subscription.go SubscriptionConfig describes the configuration of a subscription.

type Subscription

type Subscription struct {
	*pubsub.PubSub

	// Settings for receiving messages. All changes must be made before the
	// first call to Receive. The default is DefaultPublishSettings.
	// it means could not dynamically change and hot start.
	ReceiveSettings
	// contains filtered or unexported fields
}

func NewSubscription

func NewSubscription(topicName string, driverMetadata protocol.Metadata, options ...Option) (*Subscription, error)

new a topic and init it with the connection options

func (*Subscription) Receive

func (s *Subscription) Receive(ctx context.Context, r *protocol.SubscribeRequest, callback func(ctx context.Context, message *protocol.Message)) error

todo: add batching iterator to batch every suber's protocol. that's need to store the messages in subscribers. Receive is a blocking function and return error until receive the protocol and occurs error when handle protocol. if error, may should call DrainAck()?