Shimmed out Stream and Client

Author: AndrewBurian

client.go

@@ -0,0 +1,35 @@
+package eventsource
+
+import (
+	"io"
+	"net/http"
+)
+
+type Client struct {
+	flush http.Flusher
+	write io.Writer
+}
+
+func NewClient(w http.ResponseWriter) *Client {
+	c := &Client{
+		write: w,
+	}
+
+	// Check to ensure we support flushing
+	flush, ok := w.(http.Flusher)
+	if !ok {
+		return nil
+	}
+	c.flush = flush
+
+	return c
+}
+
+func (c *Client) Send(ev *Event) {
+	io.Copy(c.write, ev)
+	c.flush.Flush()
+}
+
+func (c *Client) run() {
+
+}

event.go

@@ -2,6 +2,8 @@ package eventsource
 
 import (
 	"bytes"
+	"io"
+	"io/ioutil"
 	"strconv"
 )
 
@@ -15,6 +17,18 @@ type Event struct {
 	bufSet bool
 }
 
+// DataEvent creates a new Event with the data set
+func DataEvent(data string) *Event {
+	e := &Event{}
+	io.WriteString(e, data)
+	return e
+}
+
+// ID sets the event ID
+func (e *Event) ID(id string) {
+	e.id = []byte(id)
+}
+
 // Read the event in wire format
 func (e *Event) Read(p []byte) (int, error) {
 	if e.bufSet {
@@ -80,3 +94,17 @@ func (e *Event) Write(p []byte) (int, error) {
 	e.bufSet = false
 	return len(p), nil
 }
+
+// WriteRaw sets an event directly in wire format
+//
+// This does no validation to ensure it is in a correct format
+// and should mostly be used to deep copy another event
+func (e *Event) WriteRaw(p []byte) (int, error) {
+	return e.buf.Write(p)
+}
+
+// String returns the Event in wire format as a string
+func (e *Event) String() string {
+	fullEvent, _ := ioutil.ReadAll(e)
+	return string(fullEvent)
+}

stream.go

@@ -0,0 +1,83 @@
+package eventsource
+
+import (
+	"container/list"
+	"net/http"
+	"sync"
+)
+
+type Stream struct {
+	clients           list.List
+	broadcast         chan *Event
+	shutdownWait      sync.WaitGroup
+	clientConnectHook func(*http.Request, *Client)
+}
+
+func New() *Stream {
+	s := &Stream{}
+	go s.run()
+	return s
+}
+
+func (s *Stream) Register(c *Client) {
+
+}
+
+func (s *Stream) Broadcast(e *Event) {
+
+}
+
+func (s *Stream) Subscribe(topic string, c *Client) {
+
+}
+
+func (s *Stream) Publish(topic string, e *Event) {
+
+}
+
+func (s *Stream) Shutdown() {
+
+}
+
+func (s *Stream) CloseTopic(topic string) {
+
+}
+
+func (s *Stream) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+
+}
+
+func (s *Stream) TopicHandler(topic string) http.HandlerFunc {
+
+}
+
+// Register a function to be called when a client connects to this stream's
+// HTTP handler
+func (s *Stream) ClientConnectHook(fn func(*http.Request, *Client)) {
+	s.clientConnectHook = fn
+}
+
+func (s *Stream) run() {
+
+	for {
+		select {
+
+		case ev, ok := <-s.broadcast:
+
+			// end of the broadcast channel indicates a shutdown
+			if !ok {
+				//s.closeAll()
+				s.shutdownWait.Done()
+				return
+			}
+
+			// otherwise normal message
+			s.sendAll(ev)
+
+		}
+	}
+}
+
+func (s *Stream) sendAll(ev *Event) {
+
+}