Permalink
9dce85b Jun 30, 2013
3 contributors

Users who have contributed to this file

@robfig @ShaneWilton @gabrielmiller
115 lines (96 sloc) 2.43 KB
package chatroom
import (
"container/list"
"time"
)
type Event struct {
Type string // "join", "leave", or "message"
User string
Timestamp int // Unix timestamp (secs)
Text string // What the user said (if Type == "message")
}
type Subscription struct {
Archive []Event // All the events from the archive.
New <-chan Event // New events coming in.
}
// Owner of a subscription must cancel it when they stop listening to events.
func (s Subscription) Cancel() {
unsubscribe <- s.New // Unsubscribe the channel.
drain(s.New) // Drain it, just in case there was a pending publish.
}
func newEvent(typ, user, msg string) Event {
return Event{typ, user, int(time.Now().Unix()), msg}
}
func Subscribe() Subscription {
resp := make(chan Subscription)
subscribe <- resp
return <-resp
}
func Join(user string) {
publish <- newEvent("join", user, "")
}
func Say(user, message string) {
publish <- newEvent("message", user, message)
}
func Leave(user string) {
publish <- newEvent("leave", user, "")
}
const archiveSize = 10
var (
// Send a channel here to get room events back. It will send the entire
// archive initially, and then new messages as they come in.
subscribe = make(chan (chan<- Subscription), 10)
// Send a channel here to unsubscribe.
unsubscribe = make(chan (<-chan Event), 10)
// Send events here to publish them.
publish = make(chan Event, 10)
)
// This function loops forever, handling the chat room pubsub
func chatroom() {
archive := list.New()
subscribers := list.New()
for {
select {
case ch := <-subscribe:
var events []Event
for e := archive.Front(); e != nil; e = e.Next() {
events = append(events, e.Value.(Event))
}
subscriber := make(chan Event, 10)
subscribers.PushBack(subscriber)
ch <- Subscription{events, subscriber}
case event := <-publish:
for ch := subscribers.Front(); ch != nil; ch = ch.Next() {
ch.Value.(chan Event) <- event
}
if archive.Len() >= archiveSize {
archive.Remove(archive.Front())
}
archive.PushBack(event)
case unsub := <-unsubscribe:
for ch := subscribers.Front(); ch != nil; ch = ch.Next() {
if ch.Value.(chan Event) == unsub {
subscribers.Remove(ch)
break
}
}
}
}
}
func init() {
go chatroom()
}
// Helpers
// Drains a given channel of any messages.
func drain(ch <-chan Event) {
for {
select {
case _, ok := <-ch:
if !ok {
return
}
default:
return
}
}
}