Merge pull request AndrewBurian#4 from AndrewBurian/remove-list

Remove list

Author: AndrewBurian

README.md

@@ -5,7 +5,7 @@ Swiss Army Knife for SSE in Golang
 So you want to publish events to client that connect to your server?
 ```go
 func main() {
-	stream := &eventsource.Stream{}
+	stream := eventsource.NewStream()
 
 	go func(s *eventsource.Stream) {
 		for {
@@ -24,7 +24,7 @@ The `Stream` object implements an `http.Handler` for you so it can be registered
 You got it! What do you need?
 
 ## Multiplexing / Topics / Rooms / Channels
-We call them "topics" but the gist is the same. All Clients always recieve `Broadcast` events, but you can `Publish` events to a specific topic, and then only clients that have `Subscribe`d to that topic will recieve the event.
+We call them "topics" but the gist is the same. All Clients always receive `Broadcast` events, but you can `Publish` events to a specific topic, and then only clients that have `Subscribe`d to that topic will receive the event.
 
 ```go
 stream.Subscribe("weather", myClient)
@@ -35,8 +35,8 @@ stream.Broadcast(tornadoWarningEvent)
 You can also just create multiple `Stream` objects much to the same effect, then only use `Broadcast`. Streams are cheap and run no background routines, so this is a valid pattern.
 
 ```go
-weatherStream := &eventstream.Stream{}
-lotteryStream := &eventstream.Stream{}
+weatherStream := eventstream.NewStream()
+lotteryStream := eventstream.NewStream()
 
 weatherStream.Register(clientPlanningHikes)
 lotteryStream.Register(soonToBePoorClient)
@@ -49,7 +49,7 @@ lotteryStream.Broadcast(everyoneLosesEvent)
 
 Use `TopicHandler` to create another handler for that stream that will subscribe clients to topics as well as broadcasts.
 ```go
-stream := &eventsource.Stream{}
+stream := eventsource.NewStream()
 catsHandler := stream.TopicHandler([]string{"cat"})
 
 http.ListenAndServe(":8080", catsHandler)
@@ -59,10 +59,10 @@ http.ListenAndServe(":8080", catsHandler)
 Use the stream's `Register`, `Subscribe`, `Remove`, `Unsubscribe`, and `CloseTopic` functions to control which clients are registered where.
 
 ## Tell me when clients connect
-Register a callback for the `Stream` to envoke everytime a new client connects with `Stream.ClientConnectHook`. It'll give you a handle to the resulting Client and the http request that created it, letting you do whatever you please.
+Register a callback for the `Stream` to invoke every time a new client connects with `Stream.ClientConnectHook`. It'll give you a handle to the resulting Client and the http request that created it, letting you do whatever you please.
 
 ```go
-stream := &eventsource.Stream{}
+stream := eventsource.NewStream()
 stream.ClientConnectHook(func(r *http.Requset, c *eventsource.Client){
   fmt.Println("Recieved connection from", r.Host)
   fmt.Println("Hate that guy")
@@ -71,13 +71,13 @@ stream.ClientConnectHook(func(r *http.Requset, c *eventsource.Client){
 })
 ```
 
-The callback will be on the same goroutine as the incoming web request that created it, but the Client is live and functioning so it'll start recieving broadcasts and publishments immediately before your callback has returned.
+The callback will be on the same goroutine as the incoming web request that created it, but the Client is live and functioning so it'll start receiving broadcasts and publications immediately before your callback has returned.
 
 ## Graceful shutdown
-The stream's `Shutdown` command will unsubscribe and disconnect all connected clients. However the `Stream` itself is not running any background routines, and may continue to register new clients if it's still registed as an http handler.
+The stream's `Shutdown` command will unsubscribe and disconnect all connected clients. However the `Stream` itself is not running any background routines, and may continue to register new clients if it's still registered as an http handler.
 
 ## Get out of my way
-Fine! The `Stream` object is entirely convinience. It runs no background routines and does no special handling. It just adds the topics abstraction and calls `NewClient` for you when it's connected to. Feel free not to use it.
+Fine! The `Stream` object is entirely convenience. It runs no background routines and does no special handling. It just adds the topics abstraction and calls `NewClient` for you when it's connected to. Feel free not to use it.
 
 # More control of the `Client`
 You betcha.
@@ -101,12 +101,12 @@ func ServeHTTP(w http.ResponseWriter, r *http.Request) {
 Letting the http handler routine that created the Client return may cause the underlying connection to be closed by the server. Since `NewClient` does not block, use `Wait` to block until the client is shutdown.
 
 ## Shutdown the client
-The client's `Shutdown` function terminates the background routine and marks the client as closed. It does not actually sever the connection. It does unblock any routines waiting on `Wait`, which assuming the main http hander routine was waiting there, will cause the connection to close as it returns.
+The client's `Shutdown` function terminates the background routine and marks the client as closed. It does not actually sever the connection. It does unblock any routines waiting on `Wait`, which assuming the main http handler routine was waiting there, will cause the connection to close as it returns.
 
 Attempts to `Send` events to a client after it has been shutdown will result in an error
 
 # More control of Events
-Events are the most critical part of the library, and are the most versitile.
+Events are the most critical part of the library, and are the most versatile.
 
 ## Write my own events from scratch
 Events implement the `io.ReadWriter` interface so that data can be written to it from practically any source. However the `Write` interface _does not_ write an entire event in wire format. It writes the provided buffer into `data:` sections in the resulting event.
@@ -135,7 +135,7 @@ newEvent.WriteRaw(evData)             // that will work
 newEvent.AppendData("Moar")           // ... and you ruined it
 ```
 
-Use `Clone` to create a perfect deep copy that survives further mutation. Though this is less efficient in memeory.
+Use `Clone` to create a perfect deep copy that survives further mutation. Though this is less efficient in memory.
 
 ## Create events more easily
 Since you will probably be creating more than just a few events, the `EventFactory` interface and a couple helper factories and functions have been provided to speed things up.

event.go

@@ -119,7 +119,7 @@ func (e *Event) Write(p []byte) (int, error) {
 }
 
 // WriteString adds string data to the event.
-// Equivilant to calling Write([]byte(string))
+// Equivalent to calling Write([]byte(string))
 func (e *Event) WriteString(p string) {
 	// split event on newlines
 	split := strings.Split(p, "\n")

stream.go

@@ -4,7 +4,7 @@ Package eventsource is a library for dealing with server sent events in Go.
 The library attempts to make as few assumptions as possible about how your apps
 will be written, and remains as generic as possible while still being simple and useful.
 
-The three core obects to the library are Clients, Events, and Streams.
+The three core objects to the library are Clients, Events, and Streams.
 
 Client wraps an HTTP connection, and runs a worker routine to send events to the connected
 client in a thread-safe way. It gracefully handles client disconnects.
@@ -14,12 +14,12 @@ it to wire format to send. Events are not thread-safe by themselves.
 
 Stream is an abstraction for 0 or more Client connections, and adds some multiplexing and filtering
 on top of the Client. It also can act as an http.Handler to automatically register inbound client
-connections and disconnectinos.
+connections and disconnections.
 
 A quick example of a simple sever that broadcasts a "tick" event every second
 
 	func main() {
-		stream := &eventsource.Stream{}
+		stream := eventsource.NewStream()
 		go func(s *eventsource.Stream) {
 			for {
 				time.Sleep(time.Second)
@@ -33,7 +33,6 @@ A quick example of a simple sever that broadcasts a "tick" event every second
 package eventsource
 
 import (
-	"container/list"
 	"net/http"
 	"sync"
 )
@@ -43,89 +42,93 @@ import (
 // A stream also implements an http.Handler to easily register incoming
 // http requests as new clients.
 type Stream struct {
-	clients           list.List
+	clients           map[*Client]topicList
 	listLock          sync.RWMutex
 	shutdownWait      sync.WaitGroup
 	clientConnectHook func(*http.Request, *Client)
 }
 
-type registeredClient struct {
-	c      *Client
-	topics map[string]bool
+type topicList map[string]bool
+
+// NewStream creates a new stream object
+func NewStream() *Stream {
+	return &Stream{
+		clients: make(map[*Client]topicList),
+	}
 }
 
 // Register adds a client to the stream to receive all broadcast
 // messages. Has no effect if the client is already registered.
 func (s *Stream) Register(c *Client) {
+	s.listLock.Lock()
+	defer s.listLock.Unlock()
 
 	// see if the client has been registered
-	if cli := s.getClient(c); cli != nil {
+	if _, found := s.clients[c]; found {
 		return
 	}
 
 	// append new client
-	s.addClient(c)
+	s.clients[c] = make(topicList)
 }
 
 // Remove will remove a client from this stream, but not shut the client down.
 func (s *Stream) Remove(c *Client) {
 	s.listLock.Lock()
 	defer s.listLock.Unlock()
 
-	for element := s.clients.Front(); element != nil; element = element.Next() {
-		if regCli := element.Value.(*registeredClient); regCli.c == c {
-			// client found
-			s.clients.Remove(element)
-			return
-		}
-	}
+	delete(s.clients, c)
 }
 
 // Broadcast sends the event to all clients registered on this stream.
 func (s *Stream) Broadcast(e *Event) {
 	s.listLock.RLock()
 	defer s.listLock.RUnlock()
 
-	for element := s.clients.Front(); element != nil; element = element.Next() {
-		cli := element.Value.(*registeredClient)
-		cli.c.Send(e)
+	for cli := range s.clients {
+		cli.Send(e)
 	}
 }
 
 // Subscribe add the client to the list of clients receiving publications
 // to this topic. Subscribe will also Register an unregistered
 // client.
 func (s *Stream) Subscribe(topic string, c *Client) {
+	s.listLock.Lock()
+	defer s.listLock.Unlock()
+
 	// see if the client is registered
-	cli := s.getClient(c)
+	topics, found := s.clients[c]
 
 	// register if not
-	if cli == nil {
-		cli = s.addClient(c)
+	if !found {
+		topics = make(topicList)
+		s.clients[c] = topics
 	}
 
-	cli.topics[topic] = true
+	topics[topic] = true
 }
 
 // Unsubscribe removes clients from the topic, but not from broadcasts.
 func (s *Stream) Unsubscribe(topic string, c *Client) {
+	s.listLock.Lock()
+	defer s.listLock.Unlock()
 
-	cli := s.getClient(c)
-	if cli == nil {
+	topics, found := s.clients[c]
+	if !found {
 		return
 	}
-	cli.topics[topic] = false
+	topics[topic] = false
 }
 
 // Publish sends the event to clients that have subscribed to the given topic.
 func (s *Stream) Publish(topic string, e *Event) {
 	s.listLock.RLock()
 	defer s.listLock.RUnlock()
 
-	for element := s.clients.Front(); element != nil; element = element.Next() {
-		cli := element.Value.(*registeredClient)
-		if cli.topics[topic] {
-			cli.c.Send(e)
+	for cli, topics := range s.clients {
+		if topics[topic] {
+			cli.Send(e)
 		}
 	}
 }
@@ -135,12 +138,9 @@ func (s *Stream) Shutdown() {
 	s.listLock.Lock()
 	defer s.listLock.Unlock()
 
-	for element := s.clients.Front(); element != nil; {
-		cli := element.Value.(*registeredClient)
-		cli.c.Shutdown()
-		next := element.Next()
-		s.clients.Remove(element)
-		element = next
+	for client := range s.clients {
+		client.Shutdown()
+		delete(s.clients, client)
 	}
 }
 
@@ -223,32 +223,3 @@ func (s *Stream) ClientConnectHook(fn func(*http.Request, *Client)) {
 func checkRequest(r *http.Request) bool {
 	return r.Header.Get("Accept") == "text/event-stream"
 }
-
-func (s *Stream) getClient(c *Client) *registeredClient {
-	s.listLock.RLock()
-	defer s.listLock.RUnlock()
-
-	for element := s.clients.Front(); element != nil; element = element.Next() {
-		if regCli := element.Value.(*registeredClient); regCli.c == c {
-			// client found
-			return regCli
-		}
-	}
-
-	// not found
-	return nil
-}
-
-func (s *Stream) addClient(c *Client) *registeredClient {
-
-	cli := &registeredClient{
-		c:      c,
-		topics: make(map[string]bool),
-	}
-
-	s.listLock.Lock()
-	s.clients.PushBack(cli)
-	s.listLock.Unlock()
-
-	return cli
-}