git.scottworley.com Git - reliable-chat/blob - 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, either version 3 of the
   7  *  License, or (at your option) any later version.
   8  *
   9  *  This program is distributed in the hope that it will be useful,
  10  *  but WITHOUT ANY WARRANTY; without even the implied warranty of
  11  *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  12  *  GNU Affero General Public License for more details.
  13  *
  14  *  You should have received a copy of the GNU Affero General Public License
  15  *  along with this program.  If not, see <http://www.gnu.org/licenses/>.
  16  */
  17
  18 package main
  19
  20 import "container/list"
  21 import "encoding/json"
  22 import "expvar"
  23 import "flag"
  24 import "log"
  25 import "net/http"
  26 import "strconv"
  27 import "time"
  28
  29 var port = flag.Int("port", 21059, "Port to listen on")
  30
  31 var frame_count = expvar.NewInt("frame_count")
  32 var speak_count = expvar.NewInt("speak_count")
  33 var fetch_count = expvar.NewInt("fetch_count")
  34 var fetch_wait_count = expvar.NewInt("fetch_wait_count")
  35 var fetch_wake_count = expvar.NewInt("fetch_wake_count")
  36
  37 type Message struct {
  38         Time time.Time
  39         ID   string
  40         Text string
  41 }
  42
  43 type StoreRequest struct {
  44         StartTime time.Time
  45         Messages  chan<- []Message
  46 }
  47
  48 type Store struct {
  49         Add chan *Message
  50         Get chan *StoreRequest
  51 }
  52
  53 // TODO: Monotonic clock
  54
  55 func manage_store(store Store) {
  56         messages := list.New()
  57         message_count := 0
  58         max_messages := 1000
  59         waiting := list.New()
  60 main:
  61         for {
  62                 select {
  63                 case new_message, ok := <-store.Add:
  64                         if !ok {
  65                                 break main
  66                         }
  67                         speak_count.Add(1)
  68                         for waiter := waiting.Front(); waiter != nil; waiter = waiter.Next() {
  69                                 waiter.Value.(*StoreRequest).Messages <- []Message{*new_message}
  70                                 close(waiter.Value.(*StoreRequest).Messages)
  71                                 fetch_wake_count.Add(1)
  72                         }
  73                         waiting.Init()
  74                         messages.PushBack(new_message)
  75                         if message_count < max_messages {
  76                                 message_count++
  77                         } else {
  78                                 messages.Remove(messages.Front())
  79                         }
  80                 case request, ok := <-store.Get:
  81                         if !ok {
  82                                 break main
  83                         }
  84                         fetch_count.Add(1)
  85                         if messages.Back() == nil || !request.StartTime.Before(messages.Back().Value.(*Message).Time) {
  86                                 waiting.PushBack(request)
  87                                 fetch_wait_count.Add(1)
  88                         } else {
  89                                 start := messages.Back()
  90                                 response_size := 1
  91                                 if messages.Front().Value.(*Message).Time.After(request.StartTime) {
  92                                         start = messages.Front()
  93                                         response_size = message_count
  94                                 } else {
  95                                         for start.Prev().Value.(*Message).Time.After(request.StartTime) {
  96                                                 start = start.Prev()
  97                                                 response_size++
  98                                         }
  99                                 }
 100                                 response_messages := make([]Message, 0, response_size)
 101                                 for m := start; m != nil; m = m.Next() {
 102                                         response_messages = append(response_messages, *m.Value.(*Message))
 103                                 }
 104                                 request.Messages <- response_messages
 105                         }
 106                 }
 107         }
 108 }
 109
 110 func start_store() Store {
 111         store := Store{make(chan *Message, 20), make(chan *StoreRequest, 20)}
 112         go manage_store(store)
 113         return store
 114 }
 115
 116 const frame_html = `<!DOCTYPE html PUBLIC "-//W3C//DTD XHTML 1.1//EN"
 117   "http://www.w3.org/TR/xhtml11/DTD/xhtml11.dtd">
 118
 119 <html xmlns="http://www.w3.org/1999/xhtml">
 120 <head>
 121  <script type="text/javascript"><!--//--><![CDATA[//><!--
 122   var since;
 123   window.parent.postMessage('[{"Time":"2000-01-01T00:00:00.000000-00:00","ID":"/frame deprecation warning","Text":"*** You are using an old version of the client.  Please upgrade."}]', "*");
 124   function go() {
 125    var delay = 10000;
 126    var xhr = new XMLHttpRequest();
 127    xhr.onreadystatechange = function() {
 128     if (this.readyState == this.DONE) {
 129      if (this.status == 200) {
 130       var rtxt = this.responseText;
 131       if (rtxt != null) {
 132        var r = JSON.parse(rtxt);
 133        if (r != null) {
 134         window.parent.postMessage(rtxt, "*");
 135         delay = 40;
 136         if (r.length >= 1 && "Time" in r[r.length-1]) {
 137          since = r[r.length-1]["Time"];
 138         }
 139        }
 140       }
 141      }
 142      window.setTimeout(go, delay);
 143     }
 144    }
 145    var uri = "/fetch";
 146    if (since) {
 147     uri += '?since="' + since + '"';
 148    }
 149    xhr.open("GET", uri);
 150    xhr.send();
 151   }
 152   //--><!]]></script>
 153 </head>
 154 <body onload="go()">
 155 </body>
 156 </html>
 157 `
 158
 159 func start_server(store Store) {
 160         http.HandleFunc("/fetch", func(w http.ResponseWriter, r *http.Request) {
 161                 var since time.Time
 162                 url_since := r.FormValue("since")
 163                 if url_since != "" {
 164                         err := json.Unmarshal([]byte(url_since), &since)
 165                         if err != nil {
 166                                 log.Print("fetch: parse since: ", err)
 167                                 w.WriteHeader(http.StatusBadRequest)
 168                                 w.Write([]byte("Could not parse since as date"))
 169                                 return
 170                         }
 171                 }
 172                 messages_from_store := make(chan []Message, 1)
 173                 store.Get <- &StoreRequest{since, messages_from_store}
 174
 175                 json_encoded, err := json.Marshal(<-messages_from_store)
 176                 if err != nil {
 177                         log.Print("json encode: ", err)
 178                         w.WriteHeader(http.StatusInternalServerError)
 179                         return
 180                 }
 181                 w.Header().Add("Content-Type", "application/json")
 182                 w.Header().Add("Access-Control-Allow-Origin", "*")
 183                 w.Write(json_encoded)
 184         })
 185
 186         http.HandleFunc("/speak", func(w http.ResponseWriter, r *http.Request) {
 187                 store.Add <- &Message{
 188                         time.Now(),
 189                         r.FormValue("id"),
 190                         r.FormValue("text")}
 191         })
 192
 193         http.HandleFunc("/frame", func(w http.ResponseWriter, r *http.Request) {
 194                 frame_count.Add(1)
 195                 w.Write([]byte(frame_html));
 196         })
 197
 198         log.Fatal(http.ListenAndServe(":"+strconv.Itoa(*port), nil))
 199 }
 200
 201 func main() {
 202         flag.Parse()
 203         store := start_store()
 204         start_server(store)
 205 }