func (r SSERouter) Close() error {
	return r.fanOut.Close()
}
Close stops the SSERouter.
package http
import (
	"context"
	"net/http"
	"github.com/go-chi/render"
	"github.com/pkg/errors"
	"github.com/ThreeDotsLabs/watermill"
	"github.com/ThreeDotsLabs/watermill/message"
	"github.com/ThreeDotsLabs/watermill/pubsub/gochannel"
)
type StreamAdapter interface {
	InitialStreamResponse(w http.ResponseWriter, r *http.Request) (response interface{}, ok bool)
	NextStreamResponse(r *http.Request, msg *message.Message) (response interface{}, ok bool)
}
type HandleErrorFunc func(w http.ResponseWriter, r *http.Request, err error)
type defaultErrorResponse struct {
	Error string `json:"error"`
}
func DefaultErrorHandler(w http.ResponseWriter, r *http.Request, err error) {
	w.WriteHeader(500)
	render.Respond(w, r, defaultErrorResponse{Error: err.Error()})
}
type SSERouter struct {
	fanOut *gochannel.FanOut
	config SSERouterConfig
	logger watermill.LoggerAdapter
}
type SSERouterConfig struct {
	UpstreamSubscriber message.Subscriber
	ErrorHandler       HandleErrorFunc
}
func (c *SSERouterConfig) setDefaults() {
	if c.ErrorHandler == nil {
		c.ErrorHandler = DefaultErrorHandler
	}
}
func (c SSERouterConfig) validate() error {
	if c.UpstreamSubscriber == nil {
		return errors.New("upstream subscriber is nil")
	}
	return nil
}
func NewSSERouter(
	config SSERouterConfig,
	logger watermill.LoggerAdapter,
) (SSERouter, error) {
	config.setDefaults()
	if err := config.validate(); err != nil {
		return SSERouter{}, errors.Wrap(err, "invalid SSERouter config")
	}
	if logger == nil {
		logger = watermill.NopLogger{}
	}
	fanOut, err := gochannel.NewFanOut(config.UpstreamSubscriber, logger)
	if err != nil {
		return SSERouter{}, errors.Wrap(err, "could not create a FanOut")
	}
	return SSERouter{
		fanOut: fanOut,
		config: config,
		logger: logger,
	}, nil
}
func (r SSERouter) AddHandler(topic string, streamAdapter StreamAdapter) http.HandlerFunc {
	r.logger.Trace("Adding handler for topic", watermill.LogFields{
		"topic": topic,
	})
	r.fanOut.AddSubscription(topic)
	handler := sseHandler{
		subscriber:    r.fanOut,
		topic:         topic,
		streamAdapter: streamAdapter,
		config:        r.config,
		logger:        r.logger,
	}
	return handler.Handle
}
func (r SSERouter) Run(ctx context.Context) error {
	return r.fanOut.Run(ctx)
}
func (r SSERouter) Running() chan struct{} {
	return r.fanOut.Running()
}