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 var localaddress = flag.String("localaddress", "", "Local address to bind to")
  31 var max_messages = flag.Int("maxmessages", 1000, "Maximum number of messages to retain")
  32
  33 var start_time = expvar.NewInt("start_time")
  34 var speak_count = expvar.NewInt("speak_count")
  35 var fetch_count = expvar.NewInt("fetch_count")
  36 var fetch_wait_count = expvar.NewInt("fetch_wait_count")
  37 var fetch_wake_count = expvar.NewInt("fetch_wake_count")
  38 var drop_due_to_limit_count = expvar.NewInt("drop_due_to_limit_count")
  39
  40 type Message struct {
  41         Time time.Time
  42         ID   string
  43         Text string
  44 }
  45
  46 type StoreRequest struct {
  47         StartTime time.Time
  48         Messages  chan<- []Message
  49 }
  50
  51 type Store struct {
  52         Add chan *Message
  53         Get chan *StoreRequest
  54 }
  55
  56 // TODO: Monotonic clock
  57
  58 func manage_store(store Store) {
  59         messages := list.New()
  60         message_count := 0
  61         waiting := list.New()
  62 main:
  63         for {
  64                 select {
  65                 case new_message, ok := <-store.Add:
  66                         if !ok {
  67                                 break main
  68                         }
  69                         speak_count.Add(1)
  70                         for waiter := waiting.Front(); waiter != nil; waiter = waiter.Next() {
  71                                 waiter.Value.(*StoreRequest).Messages <- []Message{*new_message}
  72                                 close(waiter.Value.(*StoreRequest).Messages)
  73                                 fetch_wake_count.Add(1)
  74                         }
  75                         waiting.Init()
  76                         messages.PushBack(new_message)
  77                         if message_count < *max_messages {
  78                                 message_count++
  79                         } else {
  80                                 messages.Remove(messages.Front())
  81                                 drop_due_to_limit_count.Add(1)
  82                         }
  83                 case request, ok := <-store.Get:
  84                         if !ok {
  85                                 break main
  86                         }
  87                         fetch_count.Add(1)
  88                         if messages.Back() == nil || !request.StartTime.Before(messages.Back().Value.(*Message).Time) {
  89                                 waiting.PushBack(request)
  90                                 fetch_wait_count.Add(1)
  91                         } else {
  92                                 start := messages.Back()
  93                                 response_size := 1
  94                                 if messages.Front().Value.(*Message).Time.After(request.StartTime) {
  95                                         start = messages.Front()
  96                                         response_size = message_count
  97                                 } else {
  98                                         for start.Prev().Value.(*Message).Time.After(request.StartTime) {
  99                                                 start = start.Prev()
 100                                                 response_size++
 101                                         }
 102                                 }
 103                                 response_messages := make([]Message, 0, response_size)
 104                                 for m := start; m != nil; m = m.Next() {
 105                                         response_messages = append(response_messages, *m.Value.(*Message))
 106                                 }
 107                                 request.Messages <- response_messages
 108                         }
 109                 }
 110         }
 111 }
 112
 113 func start_store() Store {
 114         store := Store{make(chan *Message, 20), make(chan *StoreRequest, 20)}
 115         go manage_store(store)
 116         return store
 117 }
 118
 119 const robots_txt = `User-agent: *
 120 Disallow: /
 121 `
 122
 123 func start_server(store Store) {
 124         http.HandleFunc("/fetch", func(w http.ResponseWriter, r *http.Request) {
 125                 var since time.Time
 126                 url_since := r.FormValue("since")
 127                 if url_since != "" {
 128                         err := json.Unmarshal([]byte(url_since), &since)
 129                         if err != nil {
 130                                 log.Print("fetch: parse since: ", err)
 131                                 w.WriteHeader(http.StatusBadRequest)
 132                                 w.Write([]byte("Could not parse since as date"))
 133                                 return
 134                         }
 135                 }
 136                 messages_from_store := make(chan []Message, 1)
 137                 store.Get <- &StoreRequest{since, messages_from_store}
 138
 139                 json_encoded, err := json.Marshal(<-messages_from_store)
 140                 if err != nil {
 141                         log.Print("json encode: ", err)
 142                         w.WriteHeader(http.StatusInternalServerError)
 143                         return
 144                 }
 145                 w.Header().Add("Content-Type", "application/json")
 146                 w.Header().Add("Access-Control-Allow-Origin", "*")
 147                 w.Write(json_encoded)
 148         })
 149
 150         http.HandleFunc("/speak", func(w http.ResponseWriter, r *http.Request) {
 151                 store.Add <- &Message{
 152                         time.Now(),
 153                         r.FormValue("id"),
 154                         r.FormValue("text")}
 155         })
 156
 157         http.HandleFunc("/robots.txt", func(w http.ResponseWriter, r *http.Request) {
 158                 w.Write([]byte(robots_txt))
 159         })
 160
 161         log.Fatal(http.ListenAndServe(*localaddress+":"+strconv.Itoa(*port), nil))
 162 }
 163
 164 func main() {
 165         flag.Parse()
 166         start_time.Set(time.Now().UnixNano())
 167         store := start_store()
 168         start_server(store)
 169 }