]> git.scottworley.com Git - reliable-chat/blob - server/server.go
bdfbf87b06a5ab74fa6fea1ea22d77601f589f25
[reliable-chat] / server / server.go
1 /* reliable-chat - multipath chat
2 * Copyright (C) 2012 Scott Worley <sworley@chkno.net>
3 *
4 * This program is free software: you can redistribute it and/or modify
5 * it under the terms of the GNU Affero General Public License as
6 * published by the Free Software Foundation, version 3.
7 *
8 * This program is distributed in the hope that it will be useful,
9 * but WITHOUT ANY WARRANTY; without even the implied warranty of
10 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
11 * GNU Affero General Public License for more details.
12 *
13 * You should have received a copy of the GNU Affero General Public License
14 * along with this program. If not, see <http://www.gnu.org/licenses/>.
15 */
16
17 package main
18
19 import "container/list"
20 import "encoding/json"
21 import "expvar"
22 import "flag"
23 import "log"
24 import "net/http"
25 import "strconv"
26 import "time"
27
28 var port = flag.Int("port", 21059, "Port to listen on")
29 var localaddress = flag.String("localaddress", "", "Local address to bind to")
30 var max_messages = flag.Int("maxmessages", 1000, "Maximum number of messages to retain")
31
32 var start_time = expvar.NewInt("start_time")
33 var speak_count = expvar.NewInt("speak_count")
34 var fetch_count = expvar.NewInt("fetch_count")
35 var fetch_wait_count = expvar.NewInt("fetch_wait_count")
36 var fetch_wake_count = expvar.NewInt("fetch_wake_count")
37 var drop_due_to_limit_count = expvar.NewInt("drop_due_to_limit_count")
38
39 type Message struct {
40 Time time.Time
41 ID string
42 Text string
43 }
44
45 type StoreRequest struct {
46 StartTime time.Time
47 Messages chan<- []Message
48 }
49
50 type Store struct {
51 Add chan *Message
52 Get chan *StoreRequest
53 }
54
55 var monotonic_clock chan chan time.Time
56
57 const minimum_clock_increment = time.Millisecond
58
59 func start_clock() {
60 internal_monotonic_clock := make(chan chan time.Time, 1)
61 go func() {
62 last_time := time.Now()
63 main:
64 for {
65 select {
66 case request, ok := <-internal_monotonic_clock:
67 if !ok {
68 break main
69 }
70 earliest_acceptable_time := last_time.Add(minimum_clock_increment)
71 current_time := time.Now()
72 if current_time.Before(earliest_acceptable_time) {
73 current_time = earliest_acceptable_time
74 }
75 request <- current_time
76 last_time = current_time
77 }
78 }
79 }()
80 monotonic_clock = internal_monotonic_clock
81 }
82
83 func now() time.Time {
84 c := make(chan time.Time, 0)
85 monotonic_clock <- c
86 return <-c
87 }
88
89 func manage_store(store Store) {
90 messages := list.New()
91 message_count := 0
92 waiting := list.New()
93 main:
94 for {
95 select {
96 case new_message, ok := <-store.Add:
97 if !ok {
98 break main
99 }
100 speak_count.Add(1)
101 for waiter := waiting.Front(); waiter != nil; waiter = waiter.Next() {
102 waiter.Value.(*StoreRequest).Messages <- []Message{*new_message}
103 close(waiter.Value.(*StoreRequest).Messages)
104 fetch_wake_count.Add(1)
105 }
106 waiting.Init()
107 messages.PushBack(new_message)
108 if message_count < *max_messages {
109 message_count++
110 } else {
111 messages.Remove(messages.Front())
112 drop_due_to_limit_count.Add(1)
113 }
114 case request, ok := <-store.Get:
115 if !ok {
116 break main
117 }
118 fetch_count.Add(1)
119 if messages.Back() == nil || !request.StartTime.Before(messages.Back().Value.(*Message).Time) {
120 waiting.PushBack(request)
121 fetch_wait_count.Add(1)
122 } else {
123 start := messages.Back()
124 response_size := 1
125 if messages.Front().Value.(*Message).Time.After(request.StartTime) {
126 start = messages.Front()
127 response_size = message_count
128 } else {
129 for start.Prev().Value.(*Message).Time.After(request.StartTime) {
130 start = start.Prev()
131 response_size++
132 }
133 }
134 response_messages := make([]Message, 0, response_size)
135 for m := start; m != nil; m = m.Next() {
136 response_messages = append(response_messages, *m.Value.(*Message))
137 }
138 request.Messages <- response_messages
139 }
140 }
141 }
142 }
143
144 func start_store() Store {
145 store := Store{make(chan *Message, 20), make(chan *StoreRequest, 20)}
146 go manage_store(store)
147 return store
148 }
149
150 const robots_txt = `User-agent: *
151 Disallow: /
152 `
153
154 func start_server(store Store) {
155 http.HandleFunc("/fetch", func(w http.ResponseWriter, r *http.Request) {
156 var since time.Time
157 url_since := r.FormValue("since")
158 if url_since != "" {
159 err := json.Unmarshal([]byte(url_since), &since)
160 if err != nil {
161 log.Print("fetch: parse since: ", err)
162 w.WriteHeader(http.StatusBadRequest)
163 w.Write([]byte("Could not parse since as date"))
164 return
165 }
166 }
167 messages_from_store := make(chan []Message, 1)
168 store.Get <- &StoreRequest{since, messages_from_store}
169
170 json_encoded, err := json.Marshal(<-messages_from_store)
171 if err != nil {
172 log.Print("json encode: ", err)
173 w.WriteHeader(http.StatusInternalServerError)
174 return
175 }
176 w.Header().Add("Content-Type", "application/json")
177 w.Header().Add("Access-Control-Allow-Origin", "*")
178 w.Write(json_encoded)
179 })
180
181 http.HandleFunc("/speak", func(w http.ResponseWriter, r *http.Request) {
182 store.Add <- &Message{
183 now(),
184 r.FormValue("id"),
185 r.FormValue("text")}
186 w.Header().Add("Access-Control-Allow-Origin", "*")
187 })
188
189 http.HandleFunc("/robots.txt", func(w http.ResponseWriter, r *http.Request) {
190 w.Write([]byte(robots_txt))
191 })
192
193 log.Fatal(http.ListenAndServe(*localaddress+":"+strconv.Itoa(*port), nil))
194 }
195
196 func main() {
197 flag.Parse()
198 start_clock()
199 start_time.Set(now().UnixNano())
200 store := start_store()
201 start_server(store)
202 }