]> git.scottworley.com Git - reliable-chat/blob - server/server.go
7572447610372c8dd45c6c6dc063924df36bcecf
[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 main:
32 for {
33 select {
34 case new_message, ok := <-store.Add:
35 if !ok {
36 break main
37 }
38 messages.PushBack(new_message)
39 for waiter := waiting.Front(); waiter != nil; waiter = waiter.Next() {
40 waiter.Value.(*StoreRequest).Messages <- []Message{*new_message}
41 close(waiter.Value.(*StoreRequest).Messages)
42 }
43 waiting.Init()
44 if message_count < max_messages {
45 message_count++
46 } else {
47 messages.Remove(messages.Front())
48 }
49 case request, ok := <-store.Get:
50 if !ok {
51 break main
52 }
53 if messages.Back() == nil || !request.StartTime.Before(messages.Back().Value.(*Message).Time) {
54 waiting.PushBack(request)
55 } else {
56 start := messages.Back()
57 response_size := 1
58 if messages.Front().Value.(*Message).Time.After(request.StartTime) {
59 start = messages.Front()
60 response_size = message_count
61 } else {
62 for start.Prev().Value.(*Message).Time.After(request.StartTime) {
63 start = start.Prev()
64 response_size++
65 }
66 }
67 response_messages := make([]Message, 0, response_size)
68 for m := start; m != nil; m = m.Next() {
69 response_messages = append(response_messages, *m.Value.(*Message))
70 }
71 request.Messages <- response_messages
72 }
73 }
74 }
75 }
76
77 func start_store() Store {
78 store := Store{make(chan *Message, 20), make(chan *StoreRequest, 20)}
79 go manage_store(store)
80 return store
81 }
82
83 func start_server(store Store) {
84 http.HandleFunc("/fetch", func(w http.ResponseWriter, r *http.Request) {
85 var since time.Time
86 url_since := r.FormValue("since")
87 if url_since != "" {
88 err := json.Unmarshal([]byte(url_since), &since)
89 if err != nil {
90 log.Print("fetch: parse since: ", err)
91 w.WriteHeader(http.StatusBadRequest)
92 w.Write([]byte("Could not parse since as date"))
93 return
94 }
95 }
96 messages_from_store := make(chan []Message, 1)
97 store.Get <- &StoreRequest{since, messages_from_store}
98
99 json_encoded, err := json.Marshal(<-messages_from_store)
100 if err != nil {
101 log.Print("json encode: ", err)
102 w.WriteHeader(http.StatusInternalServerError)
103 return
104 }
105 w.Header().Add("Content-Type", "application/json")
106 w.Write(json_encoded)
107 })
108
109 http.HandleFunc("/speak", func(w http.ResponseWriter, r *http.Request) {
110 store.Add <- &Message{time.Now(), r.FormValue("text")}
111 })
112
113 log.Fatal(http.ListenAndServe(":8080", nil))
114 }
115
116 func main() {
117 store := start_store()
118 start_server(store)
119 }