stream

README

What is it?

go-http-stream-reader enables Go applications to consume long running HTTP request streams such as the Twitter Streaming API. It automatically reconnects in case of errors (e.g. temporary network interruption, stalled or closed connection etc.) and does so in a scalable manner that respects the back-off rules of the remote host.

Currently the back-off rules are modelled exactly on Twitters Rules but could be made configurable if needed (feel free to open an issue or send a PR).

Streams are read using the bufio Scanner, and by default expects newline (\n) delimited data. Different delimiting routines could easily be implemented, again feel free to open an issue or send a PR.

Usage

See GoDocs here for full api documentation: https://godoc.org/github.com/Diggs/go-http-stream-reader

Open a stream by using the NewStream function, passing in the remote URL to connect to. If needed, set the HttpClient, HttpRequest or Headers properties of the HttpStream to modify the connection behaviour (for example using a custom http.Client to enable self-signed certs, or a custom http.Request to use a different method than the default GET). Finally call Connect() to begin reading data from the stream using the Data channel:

s := stream.NewStream("https://userstream.twitter.com/1.1/user.json")
s.Headers = map[string]string{"Authorization":"foobar"}
s.Connect()
for {
  select {
  case data := <-s.Data:
    glog.Infof("Data: %s", string(data))
  case err := <-s.Error:
    glog.Infof("Error: %v", err)
  case <-s.Exit:
    glog.Info("Stream closed.")
    return
  }
}

To close the stream use stream.Close(), if you need your line reading routine to exit cleanly when the stream is closed, either check the status of the Data channel or ensure you have a reader on the Exit channel like above.

The Error channel can be read to receive any errors (resumable OR fatal) that occur during the lifetime of the stream. If a fatal error does occur it will be written to the Error channel before the stream gracefully closes.

Documentation

Index

• Constants
• type HttpStream
  • func NewStream(url string) *HttpStream
  • func (s *HttpStream) Close()
  • func (s *HttpStream) Connect()

Constants

const (
	// STREAM_INACTIVITY_TIMEOUT_SECONDS specifies the amount of time to wait between receiving data
	// before a stall condition is detected and the connection is backed off.
	STREAM_INACTIVITY_TIMEOUT_SECONDS int = 90
)

Types

type HttpStream

type HttpStream struct {
	// HttpClient can be set to provide a custom HTTP client, useful if URL serves a self-signed SSL cert and validation errors need to be ignored, for example.
	HttpClient *http.Client
	// HttpRequest can be set to provide a custom HTTP request, useful in cases where the default HTTP GET verb is not appropriate, for example.
	HttpRequest *http.Request
	// URL specifies the endpoint to connect to - this will be ignored if a custom HttpRequest is set.
	Url string
	// Headers to send with the request when connecting to URL - this will be ignored if a custom HttpRequest is set.
	Headers map[string]string
	// Data provides the data channel that is handed each data chunk that is read from the stream.
	Data chan []byte
	// Error can be read to be notified of any connection errors that occur during the lifetime of the stream.
	// Fatal errors will be delivered on this channel before the stream is closed permanently via Close().
	// Reading from this channel is optional, it will not block if there is no reader.
	Error chan error
	// Exit can be read to be notified when the stream has exited permanently e.g. due to Close() being called, or a fatal error occurring.
	// Reading from this channel is optional, it will not block if there is no reader.
	Exit chan bool
	// contains filtered or unexported fields
}

func NewStream

func NewStream(url string) *HttpStream

NewStream creates a new stream instance. Override any desired properties of the httpStream object before calling Connect() to begin reading data.

func (*HttpStream) Close

func (s *HttpStream) Close()

Close permanently disconnects the stream reader and cleans up all resources.

func (*HttpStream) Connect

func (s *HttpStream) Connect()

Connect to the configured URL and begin reading data.