1 files changed, 75 insertions, 0 deletions
diff --git a/server/session/session.go b/server/session/session.go
new file mode 100644
index 0000000..54b6d14
--- /dev/null
+++ b/server/session/session.go
@@ -0,0 +1,75 @@
+package session
+
+import (
+	"citrons.xyz/talk/proto"
+)
+
+type Session struct {
+	send chan<- proto.Line
+	UserId string
+	subscribedTo map[*Stream]bool
+}
+
+func NewSession(send chan<- proto.Line) *Session {
+	var s Session
+	s.send = send
+	s.subscribedTo = make(map[*Stream]bool)
+	return &s
+}
+
+func (s *Session) Event(ev proto.Command) {
+	s.send <- proto.Line {'*', "", ev}
+}
+
+func (s *Session) Subscriptions() map[*Stream]bool {
+	return s.subscribedTo
+}
+
+func (s *Session) Subscribe(to *Stream) {
+	s.subscribedTo[to] = true
+	if to.subscribers == nil {
+		to.subscribers = make(map[*Session]bool)
+	}
+	to.subscribers[s] = true
+}
+
+func (s *Session) Unsubscribe(to *Stream) {
+	delete(s.subscribedTo, to)
+	if to.subscribers != nil {
+		delete(to.subscribers, s)
+	}
+}
+
+type Stream struct {
+	subscribers map[*Session]bool
+}
+
+func (s *Stream) Subscribers() map[*Session]bool {
+	return s.subscribers
+}
+
+func (s *Stream) Event(ev proto.Command) {
+	for sub, _ := range s.subscribers {
+		sub.Event(ev)
+	}
+}
+
+func (s *Stream) UnsubscribeAll() {
+	for sub, _ := range s.subscribers {
+		sub.Unsubscribe(s)
+	}
+}
+
+type Request struct {
+	From *Session
+	RequestId string
+	Cmd proto.Command
+}
+
+func (r Request) Reply(reply proto.Command) {
+	r.From.send <- proto.Line {'!', r.RequestId, reply}
+}
+
+func (r Request) ReplyInvalid() {
+	r.Reply(proto.Fail{"invalid", "", nil}.Cmd())
+}