broker

type Ack func()

type Backend interface {
	// Authenticate should authenticate the client using the user and password
	// values and return true if the client is eligible to continue or false
	// when the broker should terminate the connection.
	Authenticate(client *Client, user, password string) (ok bool, err error)

	// Setup is called when a new client comes online and is successfully
	// authenticated. Setup should return the already stored session for the
	// supplied id or create and return a new one if it is missing or a clean
	// session is requested. If the supplied id has a zero length, a new
	// temporary session should be returned that is not stored further. The
	// backend should also close any existing clients that use the same id.
	//
	// Note: In this call the Backend may also allocate other resources and
	// set up the client for further usage as the broker will acknowledge the
	// connection when the call returns. The Terminate function is called for
	// every client that Setup has been called for.
	Setup(client *Client, id string, clean bool) (a Session, resumed bool, err error)

	// Restore is called after the client has restored packets from the session.
	//
	// The Backend should resubscribe stored subscriptions and begin with queueing
	// missed offline messages. When all offline messages have been queued the
	// client may receive online messages. Depending on the implementation, this
	// may not be required as Dequeue will already pick up offline messages.
	Restore(client *Client) error

	// Subscribe should subscribe the passed client to the specified topics and
	// store the subscription in the session. If an Ack is provided, the
	// subscription will be acknowledged when called during or after the call to
	// Subscribe.
	//
	// Incoming messages that match the supplied subscription should be added to
	// a temporary or persistent queue that is drained when Dequeue is called.
	//
	// Retained messages that match the supplied subscription should be added to
	// a temporary queue that is also drained when Dequeue is called. The messages
	// must be delivered with the retained flag set to true.
	Subscribe(client *Client, subs []packet.Subscription, ack Ack) error

	// Unsubscribe should unsubscribe the passed client from the specified topics
	// and remove the subscriptions from the session. If an Ack is provided, the
	// unsubscription will be acknowledged when called during or after the call
	// to Unsubscribe.
	Unsubscribe(client *Client, topics []string, ack Ack) error

	// Publish should forward the passed message to all other clients that hold
	// a subscription that matches the message's topic. It should also add the
	// message to all sessions that have a matching offline subscription. The
	// later may only apply to message's with a QOS greater than 0. If an Ack is
	// provided, the message will be acknowledged when called during or after
	// the call to Publish.
	//
	// If the retained flag is set, messages with a payload should replace the
	// currently retained message. Otherwise, the currently retained message
	// should be removed. The flag should be cleared before publishing the
	// message to other subscribed clients.
	Publish(client *Client, msg *packet.Message, ack Ack) error

	// Dequeue is called by the Client to obtain the next message from the queue
	// and must return either a message or an error. The backend must only return
	// no message and no error if the client's Closing channel has been closed.
	//
	// The Backend may return an Ack to receive a signal that the message is being
	// delivered under the selected qos level and is therefore safe to be deleted
	// from the queue. The Ack will be called before Dequeue is called again.
	//
	// The returned message must have a QOS set that respects the QOS set by
	// the matching subscription.
	Dequeue(client *Client) (*packet.Message, Ack, error)

	// Terminate is called when the client goes offline. Terminate should
	// unsubscribe the passed client from all previously subscribed topics. The
	// backend may also convert a clients subscriptions to offline subscriptions.
	//
	// Note: The Backend may also clean up previously allocated resources for
	// that client as the broker will close the connection when the call
	// returns.
	Terminate(client *Client) error

	// Log is called multiple times during the lifecycle of a client see LogEvent
	// for a list of all events.
	Log(event LogEvent, client *Client, pkt packet.Generic, msg *packet.Message, err error)
}

type Client struct {

	// MaximumKeepAlive may be set during Setup to enforce a maximum keep alive
	// for this client. Missing or higher intervals will be set to the specified
	// value.
	//
	// Will default to 5 minutes.
	MaximumKeepAlive time.Duration

	// ParallelPublishes may be set during Setup to control the number of
	// parallel calls to Publish a client can perform. This setting also has an
	// effect on how many incoming packets are stored in the client's session.
	//
	// Will default to 10.
	ParallelPublishes int

	// ParallelSubscribes may be set during Setup to control the number of
	// parallel calls to Subscribe and Unsubscribe a client can perform.
	//
	// Will default to 10.
	ParallelSubscribes int

	// InflightMessages may be set during Setup to control the number of
	// inflight messages from the broker to the client. This also defines how
	// many outgoing packets are stored in the client's session.
	//
	// Will default to 10.
	InflightMessages int

	// TokenTimeout sets the timeout after which the client should fail when
	// obtaining publish, subscribe and dequeue tokens in order to prevent
	// potential deadlocks.
	//
	// Will default to 30 seconds.
	TokenTimeout time.Duration

	// PacketCallback can be set to inspect packets before processing and
	// apply rate limits. To guarantee the connection lifecycle, Connect and
	// Disconnect packets are not provided to the callback.
	PacketCallback func(packet.Generic) error

	// Ref can be used by the backend to attach a custom object to the client.
	Ref interface{}
	// contains filtered or unexported fields
}

type Engine struct {
	// The Backend that will be passed to accepted clients.
	Backend Backend

	// ReadLimit defines the initial read limit.
	ReadLimit int64

	// MaxWriteDelay defines the initial max write delay.
	MaxWriteDelay time.Duration

	// ConnectTimeout defines the timeout to receive the first packet.
	ConnectTimeout time.Duration

	// OnError can be used to receive errors from the engine. If an error is
	// received the server should be restarted.
	OnError func(error)
	// contains filtered or unexported fields
}

type LogEvent string

type MemoryBackend struct {
	// The size of the session queue.
	SessionQueueSize int

	// The time after an error is returned while waiting on a killed existing
	// client to exit.
	KillTimeout time.Duration

	// Client configuration options. See broker.Client for details.
	ClientMaximumKeepAlive   time.Duration
	ClientParallelPublishes  int
	ClientParallelSubscribes int
	ClientInflightMessages   int
	ClientTokenTimeout       time.Duration

	// A map of username and passwords that grant read and write access.
	Credentials map[string]string

	// The Logger callback handles incoming log events.
	Logger func(LogEvent, *Client, packet.Generic, *packet.Message, error)
	// contains filtered or unexported fields
}

type Session interface {
	// NextID should return the next id for outgoing packets.
	NextID() packet.ID

	// SavePacket should store a packet in the session. An eventual existing
	// packet with the same id should be quietly overwritten.
	SavePacket(session.Direction, packet.Generic) error

	// LookupPacket should retrieve a packet from the session using the packet id.
	LookupPacket(session.Direction, packet.ID) (packet.Generic, error)

	// DeletePacket should remove a packet from the session. The method should
	// not return an error if no packet with the specified id does exist.
	DeletePacket(session.Direction, packet.ID) error

	// AllPackets should return all packets currently saved in the session.
	AllPackets(session.Direction) ([]packet.Generic, error)
}