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, 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 }