http

Documentation

Index

• Constants
• Variables
• func DefaultMarshalMessageFunc(url string, msg *message.Message) (*http.Request, error)
• func DefaultUnmarshalMessageFunc(topic string, req *http.Request) (*message.Message, error)
• type MarshalMessageFunc
• type Publisher
  • func NewPublisher(config PublisherConfig, logger watermill.LoggerAdapter) (*Publisher, error)
  • func (p *Publisher) Close() error
  • func (p *Publisher) Publish(topic string, messages ...*message.Message) error
• type PublisherConfig
• type Subscriber
  • func NewSubscriber(addr string, config SubscriberConfig, logger watermill.LoggerAdapter) (*Subscriber, error)
  • func (s *Subscriber) Addr() net.Addr
  • func (s *Subscriber) Close() error
  • func (s *Subscriber) StartHTTPServer() error
  • func (s *Subscriber) Subscribe(ctx context.Context, url string) (<-chan *message.Message, error)
• type SubscriberConfig
• type UnmarshalMessageFunc

Constants

const (
	HeaderUUID     = "Message-Uuid"
	HeaderMetadata = "Message-Metadata"
	ProviderName   = "http"
)

Variables

var (
	// ErrPublisherClosed happens when trying to publish to a topic while the publisher is closed or closing.
	ErrPublisherClosed = errors.New("publisher is closed")
	ErrNoMarshalFunc   = errors.New("marshal function is missing")
	ErrErrorResponse   = errors.New("server responded with error status")
)

Functions

func DefaultMarshalMessageFunc

func DefaultMarshalMessageFunc(url string, msg *message.Message) (*http.Request, error)

DefaultMarshalMessageFunc transforms the message into a HTTP POST request. It encodes the UUID and Metadata in request headers.

func DefaultUnmarshalMessageFunc

func DefaultUnmarshalMessageFunc(topic string, req *http.Request) (*message.Message, error)

DefaultUnmarshalMessageFunc retrieves the UUID and Metadata from request headers, as encoded by DefaultMarshalMessageFunc.

Types

type MarshalMessageFunc

type MarshalMessageFunc func(url string, msg *message.Message) (*http.Request, error)

MarshalMessageFunc transforms the message into a HTTP request to be sent to the specified url.

type Publisher

type Publisher struct {
	// contains filtered or unexported fields
}

func NewPublisher

func NewPublisher(config PublisherConfig, logger watermill.LoggerAdapter) (*Publisher, error)

NewPublisher creates a new Publisher. It publishes the received messages as HTTP requests. The URL, method and payload of the request are determined by the configured MarshalMessageFunc.

func (*Publisher) Close

func (p *Publisher) Close() error

func (*Publisher) Publish

func (p *Publisher) Publish(topic string, messages ...*message.Message) error

type PublisherConfig

type PublisherConfig struct {
	MarshalMessageFunc MarshalMessageFunc
	Client             *http.Client
	// if false (default), when server responds with error (>=400) to the webhook request, the response body is logged.
	DoNotLogResponseBodyOnServerError bool
}

type Subscriber

type Subscriber struct {
	// contains filtered or unexported fields
}

Subscriber can subscribe to HTTP requests and create Watermill's messages based on them.

func NewSubscriber

func NewSubscriber(addr string, config SubscriberConfig, logger watermill.LoggerAdapter) (*Subscriber, error)

NewSubscriber creates new Subscriber.

addr is TCP address to listen on

logger is Watermill's logger.

func (*Subscriber) Addr

func (s *Subscriber) Addr() net.Addr

Addr returns the server address or nil if the server isn't running.

func (*Subscriber) Close

func (s *Subscriber) Close() error

func (*Subscriber) StartHTTPServer

func (s *Subscriber) StartHTTPServer() error

StartHTTPServer starts http server. It must be called after all Subscribe calls have completed. Just like http.Server.Serve(), it returns http.ErrServerClosed after the server's been closed. https://golang.org/pkg/net/http/#Server.Serve

func (*Subscriber) Subscribe

func (s *Subscriber) Subscribe(ctx context.Context, url string) (<-chan *message.Message, error)

Subscribe adds HTTP handler which will listen in provided url for messages.

Subscribe needs to be called before `StartHTTPServer`.

When request is sent, it will wait for the `Ack`. When Ack is received 200 HTTP status wil be sent. When Nack is sent, 500 HTTP status will be sent.

type SubscriberConfig

type SubscriberConfig struct {
	Router               chi.Router
	UnmarshalMessageFunc UnmarshalMessageFunc
}

type UnmarshalMessageFunc

type UnmarshalMessageFunc func(topic string, request *http.Request) (*message.Message, error)