func (m *manager) readPump(d *device, r ReadCloser, closeOnce *sync.Once) {
	defer d.debugLog.Log(logging.MessageKey(), "readPump exiting")
	d.debugLog.Log(logging.MessageKey(), "readPump starting")
	var (
		readError error
		decoder   = wrp.NewDecoder(nil, wrp.Msgpack)
		encoder   = wrp.NewEncoder(nil, wrp.Msgpack)
	)
	defer func() {
		closeOnce.Do(func() { m.pumpClose(d, r, CloseReason{Err: readError, Text: "readerror"}) })
	}()
	for {
		messageType, data, readError := r.ReadMessage()
		if readError != nil {
			d.errorLog.Log(logging.MessageKey(), "read error", logging.ErrorKey(), readError)
			return
		}
		if messageType != websocket.BinaryMessage {
			d.errorLog.Log(logging.MessageKey(), "skipping non-binary frame", "messageType", messageType)
			continue
		}
		var (
			message = new(wrp.Message)
			event   = Event{
				Type:     MessageReceived,
				Device:   d,
				Message:  message,
				Format:   wrp.Msgpack,
				Contents: data,
			}
		)
		decoder.ResetBytes(data)
		err := decoder.Decode(message)
		if err != nil {
			d.errorLog.Log(logging.MessageKey(), "skipping malformed WRP message", logging.ErrorKey(), err)
			continue
		}
		if !m.wrpSourceIsValid(message, d) {
			d.errorLog.Log(logging.MessageKey(), "skipping WRP message with invalid source")
			continue
		}
		if len(strings.TrimSpace(message.ContentType)) == 0 {
			message.ContentType = DefaultWRPContentType
		}
		addDeviceMetadataContext(message, d.Metadata())
		if message.Type == wrp.SimpleRequestResponseMessageType {
			m.measures.RequestResponse.Add(1.0)
		}
		encoder.ResetBytes(&event.Contents)
		err = encoder.Encode(message)
		if err != nil {
			d.errorLog.Log(logging.MessageKey(), "unable to encode WRP message", logging.ErrorKey(), err)
			continue
		}
		if message.IsTransactionPart() {
			err := d.transactions.Complete(
				message.TransactionKey(),
				&Response{
					Device:   d,
					Message:  message,
					Format:   wrp.Msgpack,
					Contents: event.Contents,
				},
			)
			if err != nil {
				d.errorLog.Log(logging.MessageKey(), "Error while completing transaction", "transactionKey", message.TransactionKey(), logging.ErrorKey(), err)
				event.Type = TransactionBroken
				event.Error = err
			} else {
				event.Type = TransactionComplete
			}
		}
		m.dispatch(&event)
	}
}
readPump is the goroutine which handles the stream of WRP messages from a device.
This goroutine exits when any error occurs on the connection.
package device
import (
	"encoding/json"
	"io"
	"net/http"
	"strconv"
	"strings"
	"sync"
	"time"
	"github.com/xmidt-org/webpa-common/v2/convey"
	"github.com/xmidt-org/webpa-common/v2/convey/conveymetric"
	"github.com/go-kit/kit/log"
	"github.com/gorilla/websocket"
	"github.com/xmidt-org/webpa-common/v2/convey/conveyhttp"
	"github.com/xmidt-org/webpa-common/v2/logging"
	"github.com/xmidt-org/webpa-common/v2/xhttp"
	"github.com/xmidt-org/wrp-go/v3"
)
const MaxDevicesHeader = "X-Xmidt-Max-Devices"
const DefaultWRPContentType = "application/octet-stream"
type Connector interface {
	Connect(http.ResponseWriter, *http.Request, http.Header) (Interface, error)
	Disconnect(ID, CloseReason) bool
	DisconnectIf(func(ID) (CloseReason, bool)) int
	DisconnectAll(CloseReason) int
	GetFilter() Filter
}
type Router interface {
	Route(*Request) (*Response, error)
}
type Registry interface {
	Len() int
	Get(ID) (Interface, bool)
	VisitAll(func(Interface) bool) int
}
type Filter interface {
	AllowConnection(d Interface) (bool, MatchResult)
}
type MatchResult struct {
	Location string
	Key      string
}
type FilterFunc func(d Interface) (bool, MatchResult)
func (filter FilterFunc) AllowConnection(d Interface) (bool, MatchResult) {
	return filter(d)
}
type Manager interface {
	Connector
	Router
	Registry
}
type ManagerOption func(*manager)
func NewManager(o *Options) Manager {
	var (
		logger      = o.logger()
		debugLogger = logging.Debug(logger)
		measures    = NewMeasures(o.metricsProvider())
		wrpCheck    = o.wrpCheck()
	)
	debugLogger.Log(logging.MessageKey(), "source check configuration", "type", wrpCheck.Type)
	return &manager{
		logger:           logger,
		errorLog:         logging.Error(logger),
		debugLog:         debugLogger,
		readDeadline:     NewDeadline(o.idlePeriod(), o.now()),
		writeDeadline:    NewDeadline(o.writeTimeout(), o.now()),
		upgrader:         o.upgrader(),
		conveyTranslator: conveyhttp.NewHeaderTranslator("", nil),
		devices: newRegistry(registryOptions{
			Logger:   logger,
			Limit:    o.maxDevices(),
			Measures: measures,
		}),
		conveyHWMetric: conveymetric.NewConveyMetric(measures.Models, []conveymetric.TagLabelPair{
			{
				Tag:   "hw-model",
				Label: "model",
			},
			{
				Tag:   "fw-name",
				Label: "firmware",
			}}...),
		deviceMessageQueueSize: o.deviceMessageQueueSize(),
		pingPeriod:             o.pingPeriod(),
		listeners:             o.listeners(),
		measures:              measures,
		enforceWRPSourceCheck: wrpCheck.Type == CheckTypeEnforce,
		filter:                o.filter(),
	}
}
type manager struct {
	logger   log.Logger
	errorLog log.Logger
	debugLog log.Logger
	readDeadline     func() time.Time
	writeDeadline    func() time.Time
	upgrader         *websocket.Upgrader
	conveyTranslator conveyhttp.HeaderTranslator
	devices        *registry
	conveyHWMetric conveymetric.Interface
	deviceMessageQueueSize int
	pingPeriod             time.Duration
	listeners             []Listener
	measures              Measures
	enforceWRPSourceCheck bool
	filter Filter
}
func (m *manager) Connect(response http.ResponseWriter, request *http.Request, responseHeader http.Header) (Interface, error) {
	m.debugLog.Log(logging.MessageKey(), "device connect", "url", request.URL)
	ctx := request.Context()
	id, ok := GetID(ctx)
	if !ok {
		xhttp.WriteError(
			response,
			http.StatusInternalServerError,
			ErrorMissingDeviceNameContext,
		)
		return nil, ErrorMissingDeviceNameContext
	}
	metadata, ok := GetDeviceMetadata(ctx)
	if !ok {
		metadata = new(Metadata)
	}
	cvy, cvyErr := m.conveyTranslator.FromHeader(request.Header)
	d := newDevice(deviceOptions{
		ID:         id,
		C:          cvy,
		Compliance: convey.GetCompliance(cvyErr),
		QueueSize:  m.deviceMessageQueueSize,
		Metadata:   metadata,
		Logger:     m.logger,
	})
	if allow, matchResults := m.filter.AllowConnection(d); !allow {
		d.infoLog.Log("filter", "filter match found,", "location", matchResults.Location, "key", matchResults.Key)
		return nil, ErrorDeviceFilteredOut
	}
	if len(metadata.Claims()) < 1 {
		d.errorLog.Log(logging.MessageKey(), "missing security information")
	}
	if cvyErr == nil {
		d.infoLog.Log("convey", cvy)
	} else {
		d.errorLog.Log(logging.MessageKey(), "bad or missing convey data", logging.ErrorKey(), cvyErr)
	}
	c, err := m.upgrader.Upgrade(response, request, responseHeader)
	if err != nil {
		d.errorLog.Log(logging.MessageKey(), "failed websocket upgrade", logging.ErrorKey(), err)
		return nil, err
	}
	d.debugLog.Log(logging.MessageKey(), "websocket upgrade complete", "localAddress", c.LocalAddr().String())
	pinger, err := NewPinger(c, m.measures.Ping, []byte(d.ID()), m.writeDeadline)
	if err != nil {
		d.errorLog.Log(logging.MessageKey(), "unable to create pinger", logging.ErrorKey(), err)
		c.Close()
		return nil, err
	}
	if err := m.devices.add(d); err != nil {
		d.errorLog.Log(logging.MessageKey(), "unable to register device", logging.ErrorKey(), err)
		c.Close()
		return nil, err
	}
	event := &Event{
		Type:   Connect,
		Device: d,
	}
	if cvyErr == nil {
		bytes, err := json.Marshal(cvy)
		if err == nil {
			event.Format = wrp.JSON
			event.Contents = bytes
		} else {
			d.errorLog.Log(logging.MessageKey(), "unable to marshal the convey header", logging.ErrorKey(), err)
		}
	}
	metricClosure, err := m.conveyHWMetric.Update(cvy, "partnerid", metadata.PartnerIDClaim(), "trust", strconv.Itoa(metadata.TrustClaim()))
	if err != nil {
		d.errorLog.Log(logging.MessageKey(), "failed to update convey metrics", logging.ErrorKey(), err)
	}
	d.conveyClosure = metricClosure
	m.dispatch(event)
	SetPongHandler(c, m.measures.Pong, m.readDeadline)
	closeOnce := new(sync.Once)
	go m.readPump(d, InstrumentReader(c, d.statistics), closeOnce)
	go m.writePump(d, InstrumentWriter(c, d.statistics), pinger, closeOnce)
	return d, nil
}
func (m *manager) dispatch(e *Event) {
	for _, listener := range m.listeners {
		listener(e)
	}
}
func (m *manager) pumpClose(d *device, c io.Closer, reason CloseReason) {
	m.devices.remove(d.id, reason)
	closeError := c.Close()
	d.errorLog.Log(logging.MessageKey(), "Closed device connection",
		"closeError", closeError, "reasonError", reason.Err, "reason", reason.Text,
		"finalStatistics", d.Statistics().String())
	m.dispatch(
		&Event{
			Type:   Disconnect,
			Device: d,
		},
	)
	d.conveyClosure()
}
func (m *manager) wrpSourceIsValid(message *wrp.Message, d *device) bool {
	expectedID := d.ID()
	if len(strings.TrimSpace(message.Source)) == 0 {
		d.errorLog.Log(logging.MessageKey(), "WRP source was empty", "trustLevel", d.Metadata().TrustClaim())
		if m.enforceWRPSourceCheck {
			m.measures.WRPSourceCheck.With("outcome", "rejected", "reason", "empty").Add(1)
			return false
		}
		m.measures.WRPSourceCheck.With("outcome", "accepted", "reason", "empty").Add(1)
		return true
	}
	actualID, err := ParseID(message.Source)
	if err != nil {
		d.errorLog.Log(logging.MessageKey(), "Failed to parse ID from WRP source", "trustLevel", d.Metadata().TrustClaim())
		if m.enforceWRPSourceCheck {
			m.measures.WRPSourceCheck.With("outcome", "rejected", "reason", "parse_error").Add(1)
			return false
		}
		m.measures.WRPSourceCheck.With("outcome", "accepted", "reason", "parse_error").Add(1)
		return true
	}
	if expectedID != actualID {
		d.errorLog.Log(logging.MessageKey(), "ID in WRP source does not match device's ID", "spoofedID", actualID, "trustLevel", d.Metadata().TrustClaim())
		if m.enforceWRPSourceCheck {
			m.measures.WRPSourceCheck.With("outcome", "rejected", "reason", "id_mismatch").Add(1)
			return false
		}
		m.measures.WRPSourceCheck.With("outcome", "accepted", "reason", "id_mismatch").Add(1)
		return true
	}
	m.measures.WRPSourceCheck.With("outcome", "accepted", "reason", "id_match").Add(1)
	return true
}
func addDeviceMetadataContext(message *wrp.Message, deviceMetadata *Metadata) {
	message.PartnerIDs = []string{deviceMetadata.PartnerIDClaim()}
	if message.Type == wrp.SimpleEventMessageType {
		message.SessionID = deviceMetadata.SessionID()
	}
}