]> git.scottworley.com Git - reliable-chat/blob - server/server.go
bfe10669e7f0bd34368984dffbde1c7fe8adcd55
[reliable-chat] / server / server.go
1 package main
2
3 import "container/list"
4 import "encoding/json"
5 import "log"
6 import "net/http"
7 import "time"
8
9 type Message struct {
10 Time time.Time
11 Text string
12 }
13
14 type StoreRequest struct {
15 StartTime time.Time
16 Messages chan<- []Message
17 }
18
19 type Store struct {
20 Add chan *Message
21 Get chan *StoreRequest
22 }
23
24 // TODO: Monotonic clock
25
26 func manage_store(store Store) {
27 messages := list.New()
28 message_count := 0
29 max_messages := 1000
30 waiting := list.New()
31 for {
32 select {
33 case new_message := <-store.Add:
34 messages.PushBack(new_message)
35 for waiter := waiting.Front(); waiter != nil; waiter = waiter.Next() {
36 waiter.Value.(*StoreRequest).Messages <- []Message{*new_message}
37 close(waiter.Value.(*StoreRequest).Messages)
38 }
39 waiting.Init()
40 if message_count < max_messages {
41 message_count++
42 } else {
43 messages.Remove(messages.Front())
44 }
45 case request := <-store.Get:
46 if messages.Back() == nil || !request.StartTime.Before(messages.Back().Value.(*Message).Time) {
47 waiting.PushBack(request)
48 } else {
49 start := messages.Back()
50 response_size := 1
51 if messages.Front().Value.(*Message).Time.After(request.StartTime) {
52 start = messages.Front()
53 response_size = message_count
54 } else {
55 for start.Prev().Value.(*Message).Time.After(request.StartTime) {
56 start = start.Prev()
57 response_size++
58 }
59 }
60 response_messages := make([]Message, 0, response_size)
61 for m := start; m != nil; m = m.Next() {
62 response_messages = append(response_messages, *m.Value.(*Message))
63 }
64 request.Messages <- response_messages
65 }
66 }
67 }
68 }
69
70 func start_store() Store {
71 store := Store{make(chan *Message, 20), make(chan *StoreRequest, 20)}
72 go manage_store(store)
73 return store
74 }
75
76 func start_server(store Store) {
77 http.HandleFunc("/fetch", func(w http.ResponseWriter, r *http.Request) {
78 var since time.Time
79 url_since := r.FormValue("since")
80 if url_since != "" {
81 err := json.Unmarshal([]byte(url_since), &since)
82 if err != nil {
83 log.Print("fetch: parse since: ", err)
84 w.WriteHeader(http.StatusBadRequest)
85 w.Write([]byte("Could not parse since as date"))
86 return
87 }
88 }
89 messages_from_store := make(chan []Message, 1)
90 store.Get <- &StoreRequest{since, messages_from_store}
91
92 json_encoded, err := json.Marshal(<-messages_from_store)
93 if err != nil {
94 log.Print("json encode: ", err)
95 w.WriteHeader(http.StatusInternalServerError)
96 return
97 }
98 w.Header().Add("Content-Type", "application/json")
99 w.Write(json_encoded)
100 })
101
102 http.HandleFunc("/speak", func(w http.ResponseWriter, r *http.Request) {
103 store.Add <- &Message{time.Now(), r.FormValue("text")}
104 })
105
106 log.Fatal(http.ListenAndServe(":8080", nil))
107 }
108
109 func main() {
110 store := start_store()
111 start_server(store)
112 }