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         Text string
  22 }
  23
  24 type StoreRequest struct {
  25         StartTime time.Time
  26         Messages  chan<- []Message
  27 }
  28
  29 type Store struct {
  30         Add chan *Message
  31         Get chan *StoreRequest
  32 }
  33
  34 // TODO: Monotonic clock
  35
  36 func manage_store(store Store) {
  37         messages := list.New()
  38         message_count := 0
  39         max_messages := 1000
  40         waiting := list.New()
  41 main:
  42         for {
  43                 select {
  44                 case new_message, ok := <-store.Add:
  45                         if !ok {
  46                                 break main
  47                         }
  48                         speak_count.Add(1)
  49                         for waiter := waiting.Front(); waiter != nil; waiter = waiter.Next() {
  50                                 waiter.Value.(*StoreRequest).Messages <- []Message{*new_message}
  51                                 close(waiter.Value.(*StoreRequest).Messages)
  52                                 fetch_wake_count.Add(1)
  53                         }
  54                         waiting.Init()
  55                         messages.PushBack(new_message)
  56                         if message_count < max_messages {
  57                                 message_count++
  58                         } else {
  59                                 messages.Remove(messages.Front())
  60                         }
  61                 case request, ok := <-store.Get:
  62                         if !ok {
  63                                 break main
  64                         }
  65                         fetch_count.Add(1)
  66                         if messages.Back() == nil || !request.StartTime.Before(messages.Back().Value.(*Message).Time) {
  67                                 waiting.PushBack(request)
  68                                 fetch_wait_count.Add(1)
  69                         } else {
  70                                 start := messages.Back()
  71                                 response_size := 1
  72                                 if messages.Front().Value.(*Message).Time.After(request.StartTime) {
  73                                         start = messages.Front()
  74                                         response_size = message_count
  75                                 } else {
  76                                         for start.Prev().Value.(*Message).Time.After(request.StartTime) {
  77                                                 start = start.Prev()
  78                                                 response_size++
  79                                         }
  80                                 }
  81                                 response_messages := make([]Message, 0, response_size)
  82                                 for m := start; m != nil; m = m.Next() {
  83                                         response_messages = append(response_messages, *m.Value.(*Message))
  84                                 }
  85                                 request.Messages <- response_messages
  86                         }
  87                 }
  88         }
  89 }
  90
  91 func start_store() Store {
  92         store := Store{make(chan *Message, 20), make(chan *StoreRequest, 20)}
  93         go manage_store(store)
  94         return store
  95 }
  96
  97 func start_server(store Store) {
  98         http.HandleFunc("/fetch", func(w http.ResponseWriter, r *http.Request) {
  99                 var since time.Time
 100                 url_since := r.FormValue("since")
 101                 if url_since != "" {
 102                         err := json.Unmarshal([]byte(url_since), &since)
 103                         if err != nil {
 104                                 log.Print("fetch: parse since: ", err)
 105                                 w.WriteHeader(http.StatusBadRequest)
 106                                 w.Write([]byte("Could not parse since as date"))
 107                                 return
 108                         }
 109                 }
 110                 messages_from_store := make(chan []Message, 1)
 111                 store.Get <- &StoreRequest{since, messages_from_store}
 112
 113                 json_encoded, err := json.Marshal(<-messages_from_store)
 114                 if err != nil {
 115                         log.Print("json encode: ", err)
 116                         w.WriteHeader(http.StatusInternalServerError)
 117                         return
 118                 }
 119                 w.Header().Add("Content-Type", "application/json")
 120                 w.Write(json_encoded)
 121         })
 122
 123         http.HandleFunc("/speak", func(w http.ResponseWriter, r *http.Request) {
 124                 store.Add <- &Message{time.Now(), r.FormValue("text")}
 125         })
 126
 127         log.Fatal(http.ListenAndServe(":"+strconv.Itoa(*port), nil))
 128 }
 129
 130 func main() {
 131         store := start_store()
 132         start_server(store)
 133 }