git.scottworley.com Git - reliable-chat/blob - server/server.go

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