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