gcppubsub

Documentation

Index

• Constants
• func Dynamic(opts *cli.Options) error
• func NewClient(opts *cli.Options) (*pubsub.Client, error)
• func Read(opts *cli.Options, md *desc.MessageDescriptor) error
• func Relay(opts *cli.Options, relayCh chan interface{}, shutdownCtx context.Context) (relay.IRelayBackend, error)
• func Write(opts *cli.Options, md *desc.MessageDescriptor) error
• type GCPPubSub
  • func (g *GCPPubSub) Read() error
  • func (g *GCPPubSub) Write(ctx context.Context, value []byte) error
• type Relayer
  • func (r *Relayer) Relay() error

Constants

const (
	RetryReadInterval = 5 * time.Second
)

Functions

func Dynamic

func Dynamic(opts *cli.Options) error

Dynamic starts up a new GRPC client connected to the dProxy service and receives a stream of outbound replay messages which are then written to the message bus.

func NewClient

func NewClient(opts *cli.Options) (*pubsub.Client, error)

func Read

func Read(opts *cli.Options, md *desc.MessageDescriptor) error

func Relay

func Relay(opts *cli.Options, relayCh chan interface{}, shutdownCtx context.Context) (relay.IRelayBackend, error)

func Write

func Write(opts *cli.Options, md *desc.MessageDescriptor) error

Write is the entry point function for performing write operations in GCP PubSub.

This is where we verify that the passed args and flags combo makes sense, attempt to establish a connection, parse protobuf before finally attempting to perform the write.

Types

type GCPPubSub

type GCPPubSub struct {
	Options *cli.Options
	MsgDesc *desc.MessageDescriptor
	Client  *pubsub.Client
	// contains filtered or unexported fields
}

func (*GCPPubSub) Read

func (g *GCPPubSub) Read() error

func (*GCPPubSub) Write

func (g *GCPPubSub) Write(ctx context.Context, value []byte) error

Write is a wrapper for amqp Publish method. We wrap it so that we can mock it in tests, add logging etc.

type Relayer

type Relayer struct {
	Client      *pubsub.Client
	Options     *cli.Options
	RelayCh     chan interface{}
	ShutdownCtx context.Context
	// contains filtered or unexported fields
}

func (*Relayer) Relay

func (r *Relayer) Relay() error