Mainflux.mainflux/ws/adapter.go

// Package ws contains the domain concept definitions needed to support
// Mainflux ws adapter service functionality.
package ws

import (
	"errors"

	"github.com/mainflux/mainflux"
	broker "github.com/nats-io/go-nats"
)

var (
	// ErrFailedMessagePublish indicates that message publishing failed.
	ErrFailedMessagePublish = errors.New("failed to publish message")

	// ErrFailedSubscription indicates that client couldn't subscribe to specified channel.
	ErrFailedSubscription = errors.New("failed to subscribe to a channel")

	// ErrFailedConnection indicates that service couldn't connect to message broker.
	ErrFailedConnection = errors.New("failed to connect to message broker")
)

// Service specifies web socket service API.
type Service interface {
	mainflux.MessagePublisher

	// Subscribes to channel with specified id.
	Subscribe(uint64, Channel) error
}

// Channel is used for receiving and sending messages.
type Channel struct {
	Messages chan mainflux.RawMessage
	Closed   chan bool
}

// Close channel and stop message transfer.
func (channel Channel) Close() {
	close(channel.Messages)
	close(channel.Closed)
}

var _ Service = (*adapterService)(nil)

type adapterService struct {
	pubsub Service
}

// New instantiates the WS adapter implementation.
func New(pubsub Service) Service {
	return &adapterService{pubsub}
}

func (as *adapterService) Publish(msg mainflux.RawMessage) error {
	if err := as.pubsub.Publish(msg); err != nil {
		switch err {
		case broker.ErrConnectionClosed, broker.ErrInvalidConnection:
			return ErrFailedConnection
		default:
			return ErrFailedMessagePublish
		}
	}
	return nil
}

func (as *adapterService) Subscribe(chanID uint64, channel Channel) error {
	if err := as.pubsub.Subscribe(chanID, channel); err != nil {
		return ErrFailedSubscription
	}
	return nil
}