golongpoll

README

golongpoll

Golang long polling library. Makes web pub-sub easy via HTTP long-poll servers and clients.

Resources

• go docs
• examples - shows how to use the longpoll server and go/js clients.
• go client
• javascript client
• longpoll http api - in case you want to create your own client.
• longpoll with gin apps

QuickStart

To create a longpoll server:

import (
  "github.com/jcuga/golongpoll"
)

// This uses the default/empty options. See section on customizing, and Options go docs.
manager, err := golongpoll.StartLongpoll(golongpoll.Options{})

// Expose pub-sub. You could omit the publish handler if you don't want
// to allow clients to publish. For example, if clients only subscribe to data.
if err == nil {
  http.HandleFunc("/events", manager.SubscriptionHandler)
  http.HandleFunc("/publish", manager.PublishHandler)
  http.ListenAndServe("127.0.0.1:8101", nil)
} else {
  // handle error creating longpoll manager--typically this means a bad option.
}

The above snippet will create a LongpollManager which has a SubscriptionHandler and a PublishHandler that can served via http (or using https). When created, the manager spins up a separate goroutine that handles the plumbing for pub-sub.

LongpollManager also has a Publish function that can be used to publish events. You can call manager.Publish("some-category", "some data here") and/or expose the manager.PublishHandler and allow publishing of events via the longpoll http api. For publishing within the same program as the manager/server, calling manager.Publish() does not use networking--under the hood it uses the go channels that are part of the pub-sub plumbing. You could also wrap the manager in an http handler closure that calls publish as desired.

See the Examples on how to use the golang and javascript clients as well as how to wrap the manager.PublishHandler or call manager.Publish() directly.

How it Works

You can think of the longpoll manager as a goroutine that uses channels to service pub-sub requests. The manager has a map[string]eventBuffer (actually a map[string]*expiringBuffer) that holds events per category as well as a data structure (another sort of map) for the subscription request book-keeping. The PublishHandler and SubscribeHandler interact with the manager goroutine via channels.

The events are stored using in-memory buffers that have a configured max number of events per category. Optionally, the events can be automatically removed based on a time-to-live setting. Since this is all in-memory, there is an optional add-on for auto-persisting and repopulating data from disk. This allows events to persist across program restarts (not the default option/behavior!) One can also create their own custom add-on as well. See the Customizing section.

One important limitation/design-decision to be aware of: the SubscriptionHandler supports subscribing to a single cateogry. If you want to subscribe to more than one category, you must make more than one call to the subscription handler--or create multiple clients each with a different category. Note however that clients are free to publish to more than one categor--to any category really, unless the manager's publish handler is not being served or there is wrapping handler logic that forbids this. Whether or not this limitation is a big deal depends on how you are using categories. This decision reduces the internal complexity and is likely not to change any time soon.

Customizing

See golongpoll.Options on how to configure the longpoll manager. This includes:

• MaxEventBufferSize - for the max number of events per category, after which oldest-first is truncated. Defaults to 250.
• EventTimeToLiveSeconds - how long events exist in the buffer, defaults to forever (as long as MaxEventBufferSize isn't reached).
• AddOn - optional way to provide custom behavior. The only add-on at the moment is FilePersistorAddOn (Usage example). See AddOn interface for creating your own custom add-on.

Remember, you don't have to expose LongpollManager.SubscriptionHandler and PublishHandler directly (or at all). You can wrap them with your own http handler that adds additional logic or validation before invoking the inner handler. See the authentication example for how to require auth via header data before those handlers get called. For publishing, you can also call manager.Publish() directly, or wrap the manager via a closure to create a custom http handler that publishes data.

long polling and gin

Need to add long polling to a Gin HTTP Framework app? Simply wrap golongpoll's manager pub/sub functions with a gin.Context and pass to router.POST and router.GET:

package main

import (
	"net/http"

	"github.com/gin-gonic/gin"

	"github.com/jcuga/golongpoll"
)

func main() {
	// Create longpoll manger with default opts
	manager, err := golongpoll.StartLongpoll(golongpoll.Options{})
	if err != nil {
		panic(err)
	}

	router := gin.Default()
	router.POST("/pub", wrapWithContext(manager.PublishHandler))
	router.GET("/sub", wrapWithContext(manager.SubscriptionHandler))
	router.Run(":8001")
}

func wrapWithContext(lpHandler func(http.ResponseWriter, *http.Request)) func(*gin.Context) {
	return func(c *gin.Context) {
		lpHandler(c.Writer, c.Request)
	}
}

Documentation

Index

• type AddOn
• type Event
• type FilePersistorAddOn
  • func NewFilePersistor(filename string, writeBufferSize int, writeFlushPeriodSeconds int) (*FilePersistorAddOn, error)
  • func (fp *FilePersistorAddOn) OnLongpollStart() <-chan *Event
  • func (fp *FilePersistorAddOn) OnPublish(event *Event)
  • func (fp *FilePersistorAddOn) OnShutdown()
• type LongpollManager
  • func StartLongpoll(opts Options) (*LongpollManager, error)
  • func (m *LongpollManager) Publish(category string, data interface{}) error
  • func (m *LongpollManager) Shutdown()
  • func (m *LongpollManager) ShutdownWithTimeout(seconds int) error
• type Options
• type PublishData
• type TrivialAddOn
  • func (tad *TrivialAddOn) OnLongpollStart() <-chan *Event
  • func (tad *TrivialAddOn) OnPublish(event *Event)
  • func (tad *TrivialAddOn) OnShutdown()

Types

type AddOn

type AddOn interface {
	// OnLongpollStart is called from StartLongpoll() before LongpollManager's
	// run goroutine starts. This should return a channel of events to
	// pre-populate within the manager. If Options.EventTimeToLiveSeconds
	// is set, then events that already too old will be skipped.
	// AddOn implementers can provide their own pre-filtering of events
	// based on the TTL for efficiency to avoid sending expired events on
	// this channel in the first place.
	//
	// The AddOn must close the returned channel when it is done sending
	// initial events to it as the LongpollManager will read from the
	// channel until it is closed. The LongpollManager's main goroutine
	// will not launch until after all data is read from this channel.
	// NOTE: if an AddOn does not wish to pre-populate any events, then
	// simply return an empty channel that is already closed.
	//
	// IMPORTANT: the events sent to the returned channel must be in
	// chronological order! The longpoll manager will panic if events are sent
	// out of order. The reason for this is that the longpoll manager
	// does not have to sort published events internally since publishing
	// happens in real time. The fact that events from Publish naturally
	// have chronological timestamps allows for many efficiencies and
	// simplifications.
	OnLongpollStart() <-chan *Event

	// OnPublish will be called with any event published by the LongpollManager.
	// Note that the manager blocks on this function completing, so if
	// an AddOn is going to perform any slow operations like a file write,
	// database insert, or a network call, then it should do so it a separate
	// goroutine to avoid blocking the manager. See FilePersistorAddOn for an
	// example of how OnPublish can do its work in its own goroutine.
	OnPublish(*Event)

	// OnShutdown will be called on LongpollManager.Shutdown().
	// AddOns can perform any cleanup or flushing before exit.
	// The LongpollManager will block on this function during shutdown.
	// Example use: FilePersistorAddOn will flush any buffered file write data
	// to disk on shutdown.
	OnShutdown()
}

AddOn provides a way to add behavior to longpolling. For example: FilePersistorAddOn in addons/persistence/file.go provides a way to persist events to file to reuse across LongpollManager runs.

type Event

type Event struct {
	// Timestamp is milliseconds since epoch to match javascrits Date.getTime().
	// This is the timestamp when the event was published.
	Timestamp int64 `json:"timestamp"`
	// Category this event belongs to. Clients subscribe to a given category.
	Category string `json:"category"`
	// Event data payload.
	// NOTE: Data can be anything that is able to passed to json.Marshal()
	Data interface{} `json:"data"`
	// Event ID, used in conjunction with Timestamp to get a complete timeline
	// of event data as there could be more than one event with the same timestamp.
	ID uuid.UUID `json:"id"`
}

Event is a longpoll event. This type has a Timestamp as milliseconds since epoch (UTC), a string category, and an arbitrary Data payload. The category is the subscription category/topic that clients can listen for via longpolling. The Data payload can be anything that is JSON serializable via the encoding/json library's json.Marshal function.

type FilePersistorAddOn

type FilePersistorAddOn struct {
	// contains filtered or unexported fields
}

FilePersistorAddOn implements the AddOn interface to provide file persistence for longpoll events. Use NewFilePersistor(string, int, int) to create a configured FilePersistorAddOn.

NOTE: uses bufio.NewScanner which has a default max scanner buffer size of 64kb (65536). So any event whose JSON is that large will fail to be read back in completely. These events, and any other events whose JSON fails to unmarshal will be skipped.

func NewFilePersistor

func NewFilePersistor(filename string, writeBufferSize int, writeFlushPeriodSeconds int) (*FilePersistorAddOn, error)

NewFilePersistor creates a new FilePersistorAddOn with the provided options. filename is the file to use for storing event data. writeBufferSize is how large a buffer is used to buffer output before writing to file. This is simply the underlying bufio.Writer's buffer size. writeFlushPeriodSeconds is how often to flush buffer to disk even when the output buffer is not filled completely. This helps avoid data loss in the event of a program crash. Returns a new FilePersistorAddOn and nil error on success, or a nil addon and a non-nil error explaining what went wrong creating a new addon.

func (*FilePersistorAddOn) OnLongpollStart

func (fp *FilePersistorAddOn) OnLongpollStart() <-chan *Event

OnLongpollStart returns a channel of events that is populated via a separate goroutine so that the calling LongpollManager can begin consuming events while we're still reading more events in from file. Note that the goroutine that adds events to this returned channel will call close() on the channel when it is out of initial events. The LongpollManager will wait for more events until the channel is closed.

func (*FilePersistorAddOn) OnPublish

func (fp *FilePersistorAddOn) OnPublish(event *Event)

OnPublish will write new events to file. Events are sent via channel to a separate goroutine so that the calling LongpollManager does not block on the file writing.

func (*FilePersistorAddOn) OnShutdown

func (fp *FilePersistorAddOn) OnShutdown()

OnShutdown will signal the run goroutine to flush any remaining event data to disk and wait for it to complete.

type LongpollManager

type LongpollManager struct {

	// SubscriptionHandler is an Http handler function that can be served
	// directly or wrapped within another handler function that adds additional
	// behavior like authentication or business logic.
	SubscriptionHandler func(w http.ResponseWriter, r *http.Request)
	// PublishHandler is an Http handler function that can be served directly
	// or wrapped within another handler function that adds additional
	// behavior like authentication or business logic. If one does not want
	// to expose the PublishHandler, and instead only publish via
	// LongpollManager.Publish(), then simply don't serve this handler.
	// NOTE: this default publish handler does not enforce anything like what
	// category or data is allowed. It is entirely permissive by default.
	// For more production-type uses, wrap this with a http handler that
	// limits what can be published.
	PublishHandler func(w http.ResponseWriter, r *http.Request)
	// contains filtered or unexported fields
}

LongpollManager is used to interact with the internal longpolling pup-sub goroutine that is launched via StartLongpoll(Options).

LongpollManager.SubscriptionHandler can be served directly or wrapped in an Http handler to add custom behavior like authentication. Events can be published via LongpollManager.Publish(). The manager can be stopped via Shutdown() or ShutdownWithTimeout(seconds int).

Events can also be published by http clients via LongpollManager.PublishHandler if desired. Simply serve this handler directly, or wrapped in an Http handler that adds desired authentication/access controls.

If for some reason you want multiple goroutines handling different pub-sub channels, you can simply create multiple LongpollManagers and serve their subscription handlers on separate URLs.

func StartLongpoll

func StartLongpoll(opts Options) (*LongpollManager, error)

StartLongpoll creates a LongpollManager, starts the internal pub-sub goroutine and returns the manager reference which you can use anywhere to Publish() events or attach a URL to the manager's SubscriptionHandler member. This function takes an Options struct that configures the longpoll behavior. If Options.EventTimeToLiveSeconds is omitted, the default is forever.

func (*LongpollManager) Publish

func (m *LongpollManager) Publish(category string, data interface{}) error

Publish an event for a given subscription category. This event can have any arbitrary data that is convert-able to JSON via the standard's json.Marshal() the category param must be a non-empty string no longer than 1024, otherwise you get an error. Cannot be called after LongpollManager.Shutdown() or LongpollManager.ShutdownWithTimeout(seconds int).

func (*LongpollManager) Shutdown

func (m *LongpollManager) Shutdown()

Shutdown will stop the LongpollManager's run goroutine and call Addon.OnShutdown. This will block on the Addon's shutdown call if an AddOn is provided. In addition to allowing a graceful shutdown, this can be useful if you want to turn off longpolling without terminating your program. After a shutdown, you can't call Publish() or get any new results from the SubscriptionHandler. Multiple calls to this function on the same manager will result in a panic.

func (*LongpollManager) ShutdownWithTimeout

func (m *LongpollManager) ShutdownWithTimeout(seconds int) error

ShutdownWithTimeout will call Shutdown but only block for a provided amount of time when waiting for the shutdown to complete. Returns an error on timeout, otherwise nil. This can only be called once otherwise it will panic.

type Options

type Options struct {
	// Whether or not to print non-error logs about longpolling.
	// Useful mainly for debugging, defaults to false.
	// NOTE: this will log every event's contents which can be spammy!
	LoggingEnabled bool

	// Max client timeout seconds to be accepted by the SubscriptionHandler
	// (The 'timeout' HTTP query param).  Defaults to 110.
	// NOTE: if serving behind a proxy/webserver, make sure the max allowed
	// timeout here is less than that server's configured HTTP timeout!
	// Typically, servers will have a 60 or 120 second timeout by default.
	MaxLongpollTimeoutSeconds int

	// How many events to buffer per subscriptoin category before discarding
	// oldest events due to buffer being exhausted.  Larger buffer sizes are
	// useful for high volumes of events in the same categories.  But for
	// low-volumes, smaller buffer sizes are more efficient.  Defaults to 250.
	MaxEventBufferSize int

	// How long (seconds) events remain in their respective category's
	// eventBuffer before being deleted. Deletes old events even if buffer has
	// the room.  Useful to save space if you don't need old events.
	// You can use a large MaxEventBufferSize to handle spikes in event volumes
	// in a single category but have a relatively short EventTimeToLiveSeconds
	// value to save space in the more common low-volume case.
	// Defaults to infinite/forever TTL.
	EventTimeToLiveSeconds int

	// Whether or not to delete an event as soon as it is retrieved via an
	// HTTP longpoll.  Saves on space if clients only interested in seeing an
	// event once and never again.  Meant mostly for scenarios where events
	// act as a sort of notification and each subscription category is assigned
	// to a single client.  As soon as any client(s) pull down this event, it's
	// gone forever.  Notice how multiple clients can get the event if there
	// are multiple clients actively in the middle of a longpoll when a new
	// event occurs.  This event gets sent to all listening clients and then
	// the event skips being placed in a buffer and is gone forever.
	DeleteEventAfterFirstRetrieval bool

	// Optional add-on to add behavior like event persistence to longpolling.
	AddOn AddOn
}

Options for LongpollManager that get sent to StartLongpoll(options)

type PublishData

type PublishData struct {
	Category string      `json:"category"`
	Data     interface{} `json:"data"`
}

PublishData is the json data that LongpollManager.PublishHandler expects.

type TrivialAddOn

type TrivialAddOn struct{}

TrivialAddOn is a trivial implementation of the AddOn interface. AddOn implementers can embed the TrivialAddOn if they only care to implement some of the required functions.

For example, if you wanted an addon that simply logged when an event is published, you could do:

type LoggerAddOn struct {
  TrivialAddOn
}

func (lad *LoggerAddOn) OnPublish(event *Event) {
  log.Printf("Event was published: %v\n", event)
}

func (*TrivialAddOn) OnLongpollStart

func (tad *TrivialAddOn) OnLongpollStart() <-chan *Event

OnLongpollStart returns an empty, closed events channel.

func (*TrivialAddOn) OnPublish

func (tad *TrivialAddOn) OnPublish(event *Event)

OnPublish does nothing

func (*TrivialAddOn) OnShutdown

func (tad *TrivialAddOn) OnShutdown()

OnShutdown does nothing