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