message.go

package analytics

import (
	"encoding/json"
	"time"
)

// Callback interface is used by analytics clients to notify
// the application when a message send succeeded or failed.
//
// Callback methods are called by a client's internal goroutines, there are no
// guarantees on which goroutine will trigger the callbacks, the calls can be
// made sequentially or in parallel, the order doesn't depend on the order of
// messages were queued to the client.
//
// Callback methods must return quickly and not cause long blocking operations
// to avoid interferring with the client's internal work flow.
type Callback interface {

	// This method is called for every message that was successfully sent to
	// the API.
	Success(Message)

	// This method is called for every message that failed to be sent to the
	// API and will be discarded by the client.
	Failure(Message, error)
}

// Message interface is used to represent analytics objects that can be sent via
// a client.
//
// Types like analytics.Track, analytics.Page, etc... implement this interface
// and therefore can be passed to the analytics.Client.Send method.
type Message interface {

	// Returns tags associated with the message.
	tags() []string

	// Validates the internal structure of the message, the method must return
	// nil if the message is valid, or an error describing what went wrong.
	validate() error
}

// Takes a message id as first argument and returns it, unless it's the zero-
// value, in that case the default id passed as second argument is returned.
func makeMessageID(id string, def string) string {
	if len(id) == 0 {
		return def
	}
	return id
}

// Returns the time value passed as first argument, unless it's the zero-value,
// in that case the default value passed as second argument is returned.
func makeTimestamp(t Time, def Time) Time {
	if t == (Time(time.Time{})) {
		return def
	}
	return t
}

// This structure represents objects sent to the /v1/batch endpoint. We don't
// export this type because it's only meant to be used internally to send groups
// of messages in one API call.
type batch struct {
	MessageID string    `json:"messageId"`
	SentAt    Time      `json:"sentAt"`
	Messages  []message `json:"batch"`
	Context   *Context  `json:"context"`
}

type serializedMessage struct {
	msg  Message
	json []byte
}

func (m *serializedMessage) MarshalJSON() ([]byte, error) {
	return m.json, nil
}

func (m *serializedMessage) Msg() Message {
	return m.msg
}

func (m *serializedMessage) size() int {
	// The `+ 1` is for the comma that sits between each items of a JSON array.
	return len(m.json) + 1
}

type message interface {
	MarshalJSON() ([]byte, error)
	Msg() Message
	size() int
}

func makeMessage(m Message, maxBytes int) (message, error) {
	result := &serializedMessage{msg: m}
	b, err := json.Marshal(m)
	if err != nil {
		return result, err
	}
	if len(b) > maxBytes {
		return result, ErrMessageTooBig
	}
	result.json = b
	return result, nil
}

type messageQueue struct {
	pending       []message
	bytes         int
	maxBatchSize  int
	maxBatchBytes int
}

func (q *messageQueue) push(m message) (b []message) {
	if (q.bytes + m.size()) > q.maxBatchBytes {
		b = q.flush()
	}

	if q.pending == nil {
		q.pending = make([]message, 0, q.maxBatchSize)
	}

	q.pending = append(q.pending, m)
	q.bytes += m.size()

	if b == nil && len(q.pending) == q.maxBatchSize {
		b = q.flush()
	}

	return
}

func (q *messageQueue) flush() (msgs []message) {
	msgs, q.pending, q.bytes = q.pending, nil, 0
	return
}

const (
	maxBatchBytes   = 500000
	maxMessageBytes = 15000
)