git.scottworley.com Git - reliable-chat/blob - server/server.go

   1 package main
   2
   3 import "container/list"
   4 import "encoding/json"
   5 import "expvar"
   6 import "flag"
   7 import "log"
   8 import "net/http"
   9 import "strconv"
  10 import "time"
  11
  12 var port = flag.Int("port", 21059, "Port to listen on")
  13
  14 var speak_count = expvar.NewInt("speak_count")
  15 var fetch_count = expvar.NewInt("fetch_count")
  16 var fetch_wait_count = expvar.NewInt("fetch_wait_count")
  17 var fetch_wake_count = expvar.NewInt("fetch_wake_count")
  18
  19 type Message struct {
  20         Time time.Time
  21         ID   string
  22         Text string
  23 }
  24
  25 type StoreRequest struct {
  26         StartTime time.Time
  27         Messages  chan<- []Message
  28 }
  29
  30 type Store struct {
  31         Add chan *Message
  32         Get chan *StoreRequest
  33 }
  34
  35 // TODO: Monotonic clock
  36
  37 func manage_store(store Store) {
  38         messages := list.New()
  39         message_count := 0
  40         max_messages := 1000
  41         waiting := list.New()
  42 main:
  43         for {
  44                 select {
  45                 case new_message, ok := <-store.Add:
  46                         if !ok {
  47                                 break main
  48                         }
  49                         speak_count.Add(1)
  50                         for waiter := waiting.Front(); waiter != nil; waiter = waiter.Next() {
  51                                 waiter.Value.(*StoreRequest).Messages <- []Message{*new_message}
  52                                 close(waiter.Value.(*StoreRequest).Messages)
  53                                 fetch_wake_count.Add(1)
  54                         }
  55                         waiting.Init()
  56                         messages.PushBack(new_message)
  57                         if message_count < max_messages {
  58                                 message_count++
  59                         } else {
  60                                 messages.Remove(messages.Front())
  61                         }
  62                 case request, ok := <-store.Get:
  63                         if !ok {
  64                                 break main
  65                         }
  66                         fetch_count.Add(1)
  67                         if messages.Back() == nil || !request.StartTime.Before(messages.Back().Value.(*Message).Time) {
  68                                 waiting.PushBack(request)
  69                                 fetch_wait_count.Add(1)
  70                         } else {
  71                                 start := messages.Back()
  72                                 response_size := 1
  73                                 if messages.Front().Value.(*Message).Time.After(request.StartTime) {
  74                                         start = messages.Front()
  75                                         response_size = message_count
  76                                 } else {
  77                                         for start.Prev().Value.(*Message).Time.After(request.StartTime) {
  78                                                 start = start.Prev()
  79                                                 response_size++
  80                                         }
  81                                 }
  82                                 response_messages := make([]Message, 0, response_size)
  83                                 for m := start; m != nil; m = m.Next() {
  84                                         response_messages = append(response_messages, *m.Value.(*Message))
  85                                 }
  86                                 request.Messages <- response_messages
  87                         }
  88                 }
  89         }
  90 }
  91
  92 func start_store() Store {
  93         store := Store{make(chan *Message, 20), make(chan *StoreRequest, 20)}
  94         go manage_store(store)
  95         return store
  96 }
  97
  98 const frame_html = `<!DOCTYPE html PUBLIC "-//W3C//DTD XHTML 1.1//EN"
  99   "http://www.w3.org/TR/xhtml11/DTD/xhtml11.dtd">
 100
 101 <html xmlns="http://www.w3.org/1999/xhtml">
 102 <head>
 103  <script type="text/javascript"><!--//--><![CDATA[//><!--
 104   var since;
 105   function go() {
 106    var delay = 10000;
 107    var xhr = new XMLHttpRequest();
 108    xhr.onreadystatechange = function() {
 109     if (this.readyState == this.DONE) {
 110      if (this.status == 200) {
 111       var rtxt = this.responseText;
 112       if (rtxt != null) {
 113        var r = JSON.parse(rtxt);
 114        if (r != null) {
 115         window.parent.postMessage(rtxt, "*");
 116         delay = 40;
 117         if (r.length >= 1 && "Time" in r[r.length-1]) {
 118          since = r[r.length-1]["Time"];
 119         }
 120        }
 121       }
 122      }
 123      window.setTimeout(go, delay);
 124     }
 125    }
 126    var uri = "/fetch";
 127    if (since) {
 128     uri += '?since="' + since + '"';
 129    }
 130    xhr.open("GET", uri);
 131    xhr.send();
 132   }
 133   //--><!]]></script>
 134 </head>
 135 <body onload="go()">
 136 </body>
 137 </html>
 138 `
 139
 140 func start_server(store Store) {
 141         http.HandleFunc("/fetch", func(w http.ResponseWriter, r *http.Request) {
 142                 var since time.Time
 143                 url_since := r.FormValue("since")
 144                 if url_since != "" {
 145                         err := json.Unmarshal([]byte(url_since), &since)
 146                         if err != nil {
 147                                 log.Print("fetch: parse since: ", err)
 148                                 w.WriteHeader(http.StatusBadRequest)
 149                                 w.Write([]byte("Could not parse since as date"))
 150                                 return
 151                         }
 152                 }
 153                 messages_from_store := make(chan []Message, 1)
 154                 store.Get <- &StoreRequest{since, messages_from_store}
 155
 156                 json_encoded, err := json.Marshal(<-messages_from_store)
 157                 if err != nil {
 158                         log.Print("json encode: ", err)
 159                         w.WriteHeader(http.StatusInternalServerError)
 160                         return
 161                 }
 162                 w.Header().Add("Content-Type", "application/json")
 163                 w.Write(json_encoded)
 164         })
 165
 166         http.HandleFunc("/speak", func(w http.ResponseWriter, r *http.Request) {
 167                 store.Add <- &Message{
 168                         time.Now(),
 169                         r.FormValue("id"),
 170                         r.FormValue("text")}
 171         })
 172
 173         http.HandleFunc("/frame", func(w http.ResponseWriter, r *http.Request) {
 174                 w.Write([]byte(frame_html));
 175         })
 176
 177         log.Fatal(http.ListenAndServe(":"+strconv.Itoa(*port), nil))
 178 }
 179
 180 func main() {
 181         flag.Parse()
 182         store := start_store()
 183         start_server(store)
 184 }