orion/emlid/client.go

package emlid

import (
	"context"
	"sync"

	"forge.cadoles.com/Pyxis/golang-socketio"
	"forge.cadoles.com/Pyxis/golang-socketio/transport"
	"github.com/mitchellh/mapstructure"
	"github.com/pkg/errors"
)

// Client is a ReachRS websocket API client
type Client struct {
	opts *Options
	conn *gosocketio.Client
}

// EventHandler is an event handler
type EventHandler func(data interface{})

// Connect connects the client to the ReachView endpoint
// This method is not safe to call by different coroutines
func (c *Client) Connect() error {

	var err error
	var wg sync.WaitGroup

	wg.Add(1)

	transport := &transport.WebsocketTransport{
		PingInterval:   c.opts.PingInterval,
		PingTimeout:    c.opts.PingTimeout,
		ReceiveTimeout: c.opts.ReceiveTimeout,
		SendTimeout:    c.opts.SendTimeout,
		BufferSize:     c.opts.BufferSize,
	}

	c.Logf("connecting to '%s'", c.opts.Endpoint)
	conn, err := gosocketio.Dial(c.opts.Endpoint, transport)
	if err != nil {
		return errors.Wrap(err, "error while connecting to endpoint")
	}

	c.conn = conn

	err = conn.On(gosocketio.OnConnection, func(h *gosocketio.Channel) {
		conn.Off(gosocketio.OnError)
		c.Logf("connected with sid '%s'", h.Id())
		err = c.sendBrowserConnected()
		wg.Done()
	})
	if err != nil {
		return errors.Wrap(err, "error while attaching to connection event")
	}

	err = conn.On(gosocketio.OnError, func(h *gosocketio.Channel) {
		c.Logf("error")
		err = errors.Errorf("an unknown error occured")
		c.conn = nil
		wg.Done()
	})
	if err != nil {
		return errors.Wrap(err, "error while attaching to error event")
	}

	wg.Wait()

	return err
}

// Close closes the current connection to the ReachView endpoint
func (c *Client) Close() {
	if c.conn == nil {
		return
	}
	c.conn.Close()
	c.conn = nil
}

// Emit a new event with the given data
func (c *Client) Emit(event string, data interface{}) error {
	c.Logf("emit '%s' event", event)
	if err := c.conn.Emit(event, data); err != nil {
		return errors.Wrapf(err, "error while emitting '%s' event", event)
	}
	c.Logf("'%s' event sent", event)
	return nil
}

// On binds and event handler to the given event
func (c *Client) On(event string, handler interface{}) error {
	return c.conn.On(event, handler)
}

// Off remove the handler bound to the given event
func (c *Client) Off(event string) {
	c.conn.Off(event)
}

// ReqResp emits an event with the given data and waits for a response
func (c *Client) ReqResp(ctx context.Context,
	requestEvent string, requestData interface{},
	responseEvent string, res interface{}) error {

	var err error
	var wg sync.WaitGroup
	var once sync.Once

	done := func() {
		c.conn.Off(responseEvent)
		wg.Done()
	}

	wg.Add(1)

	go func() {
		<-ctx.Done()
		err = ctx.Err()
		once.Do(done)
	}()

	err = c.conn.On(responseEvent, func(_ *gosocketio.Channel, data interface{}) {
		err = mapstructure.WeakDecode(data, res)
		once.Do(done)
	})
	if err != nil {
		return errors.Wrapf(err, "error while binding to '%s' event", responseEvent)
	}

	if err = c.Emit(requestEvent, requestData); err != nil {
		return errors.Wrapf(err, "error while emitting event '%s'", requestEvent)
	}

	wg.Wait()

	return err

}

// Logf logs the given message with the configured logger
func (c *Client) Logf(format string, args ...interface{}) {
	if c.opts.Logger == nil {
		return
	}
	c.opts.Logger.Printf(format, args...)
}

// NewClient returns a new ReachRS websocket client
func NewClient(opts ...OptionFunc) *Client {
	options := DefaultOptions()
	for _, o := range opts {
		o(options)
	}
	return &Client{
		opts: options,
	}
}