[reliable-chat] / server / server.go

package main

import "container/list"
import "encoding/json"
import "expvar"
import "flag"
import "log"
import "net/http"
import "strconv"
import "time"

var port = flag.Int("port", 21059, "Port to listen on")

var frame_count = expvar.NewInt("frame_count")
var speak_count = expvar.NewInt("speak_count")
var fetch_count = expvar.NewInt("fetch_count")
var fetch_wait_count = expvar.NewInt("fetch_wait_count")
var fetch_wake_count = expvar.NewInt("fetch_wake_count")

type Message struct {
	Time time.Time
	ID   string
	Text string
}

type StoreRequest struct {
	StartTime time.Time
	Messages  chan<- []Message
}

type Store struct {
	Add chan *Message
	Get chan *StoreRequest
}

// TODO: Monotonic clock

func manage_store(store Store) {
	messages := list.New()
	message_count := 0
	max_messages := 1000
	waiting := list.New()
main:
	for {
		select {
		case new_message, ok := <-store.Add:
			if !ok {
				break main
			}
			speak_count.Add(1)
			for waiter := waiting.Front(); waiter != nil; waiter = waiter.Next() {
				waiter.Value.(*StoreRequest).Messages <- []Message{*new_message}
				close(waiter.Value.(*StoreRequest).Messages)
				fetch_wake_count.Add(1)
			}
			waiting.Init()
			messages.PushBack(new_message)
			if message_count < max_messages {
				message_count++
			} else {
				messages.Remove(messages.Front())
			}
		case request, ok := <-store.Get:
			if !ok {
				break main
			}
			fetch_count.Add(1)
			if messages.Back() == nil || !request.StartTime.Before(messages.Back().Value.(*Message).Time) {
				waiting.PushBack(request)
				fetch_wait_count.Add(1)
			} else {
				start := messages.Back()
				response_size := 1
				if messages.Front().Value.(*Message).Time.After(request.StartTime) {
					start = messages.Front()
					response_size = message_count
				} else {
					for start.Prev().Value.(*Message).Time.After(request.StartTime) {
						start = start.Prev()
						response_size++
					}
				}
				response_messages := make([]Message, 0, response_size)
				for m := start; m != nil; m = m.Next() {
					response_messages = append(response_messages, *m.Value.(*Message))
				}
				request.Messages <- response_messages
			}
		}
	}
}

func start_store() Store {
	store := Store{make(chan *Message, 20), make(chan *StoreRequest, 20)}
	go manage_store(store)
	return store
}

const frame_html = `<!DOCTYPE html PUBLIC "-//W3C//DTD XHTML 1.1//EN"
  "http://www.w3.org/TR/xhtml11/DTD/xhtml11.dtd">

<html xmlns="http://www.w3.org/1999/xhtml">
<head>
 <script type="text/javascript"><!--//--><![CDATA[//><!--
  var since;
  function go() {
   var delay = 10000;
   var xhr = new XMLHttpRequest();
   xhr.onreadystatechange = function() {
    if (this.readyState == this.DONE) {
     if (this.status == 200) {
      var rtxt = this.responseText;
      if (rtxt != null) {
       var r = JSON.parse(rtxt);
       if (r != null) {
        window.parent.postMessage(rtxt, "*");
        delay = 40;
        if (r.length >= 1 && "Time" in r[r.length-1]) {
         since = r[r.length-1]["Time"];
        }
       }
      }
     }
     window.setTimeout(go, delay);
    }
   }
   var uri = "/fetch";
   if (since) {
    uri += '?since="' + since + '"';
   }
   xhr.open("GET", uri);
   xhr.send();
  }
  //--><!]]></script>
</head>
<body onload="go()">
</body>
</html>
`

func start_server(store Store) {
	http.HandleFunc("/fetch", func(w http.ResponseWriter, r *http.Request) {
		var since time.Time
		url_since := r.FormValue("since")
		if url_since != "" {
			err := json.Unmarshal([]byte(url_since), &since)
			if err != nil {
				log.Print("fetch: parse since: ", err)
				w.WriteHeader(http.StatusBadRequest)
				w.Write([]byte("Could not parse since as date"))
				return
			}
		}
		messages_from_store := make(chan []Message, 1)
		store.Get <- &StoreRequest{since, messages_from_store}

		json_encoded, err := json.Marshal(<-messages_from_store)
		if err != nil {
			log.Print("json encode: ", err)
			w.WriteHeader(http.StatusInternalServerError)
			return
		}
		w.Header().Add("Content-Type", "application/json")
		w.Write(json_encoded)
	})

	http.HandleFunc("/speak", func(w http.ResponseWriter, r *http.Request) {
		store.Add <- &Message{
			time.Now(),
			r.FormValue("id"),
			r.FormValue("text")}
	})

	http.HandleFunc("/frame", func(w http.ResponseWriter, r *http.Request) {
		frame_count.Add(1)
		w.Write([]byte(frame_html));
	})

	log.Fatal(http.ListenAndServe(":"+strconv.Itoa(*port), nil))
}

func main() {
	flag.Parse()
	store := start_store()
	start_server(store)
}
Commit	Line	Data
92d1d6ac SW	1	package main
	2
	3	import "container/list"
	4	import "encoding/json"
03513e5c	5	import "expvar"
24a546cf	6	import "flag"
92d1d6ac SW	7	import "log"
92d1d6ac SW	8	import "net/http"
24a546cf	9	import "strconv"
92d1d6ac SW	10	import "time"
92d1d6ac SW	11
24a546cf SW	12	var port = flag.Int("port", 21059, "Port to listen on")
24a546cf SW	13
a69e0831	14	var frame_count = expvar.NewInt("frame_count")
03513e5c SW	15	var speak_count = expvar.NewInt("speak_count")
	16	var fetch_count = expvar.NewInt("fetch_count")
	17	var fetch_wait_count = expvar.NewInt("fetch_wait_count")
	18	var fetch_wake_count = expvar.NewInt("fetch_wake_count")
	19
92d1d6ac SW	20	type Message struct {
92d1d6ac SW	21	Time time.Time
b199796a	22	ID string
92d1d6ac SW	23	Text string
	24	}
	25
92d1d6ac SW	26	type StoreRequest struct {
	27	StartTime time.Time
	28	Messages chan<- []Message
	29	}
	30
	31	type Store struct {
fa5e7c1b SW	32	Add chan *Message
fa5e7c1b SW	33	Get chan *StoreRequest
92d1d6ac SW	34	}
	35
	36	// TODO: Monotonic clock
	37
	38	func manage_store(store Store) {
	39	messages := list.New()
	40	message_count := 0
	41	max_messages := 1000
	42	waiting := list.New()
c282d878	43	main:
92d1d6ac SW	44	for {
92d1d6ac SW	45	select {
c282d878 SW	46	case new_message, ok := <-store.Add:
	47	if !ok {
	48	break main
	49	}
03513e5c	50	speak_count.Add(1)
92d1d6ac	51	for waiter := waiting.Front(); waiter != nil; waiter = waiter.Next() {
fa5e7c1b SW	52	waiter.Value.(StoreRequest).Messages <- []Message{new_message}
fa5e7c1b SW	53	close(waiter.Value.(*StoreRequest).Messages)
03513e5c	54	fetch_wake_count.Add(1)
92d1d6ac SW	55	}
92d1d6ac SW	56	waiting.Init()
03513e5c	57	messages.PushBack(new_message)
92d1d6ac SW	58	if message_count < max_messages {
	59	message_count++
	60	} else {
	61	messages.Remove(messages.Front())
	62	}
c282d878 SW	63	case request, ok := <-store.Get:
	64	if !ok {
	65	break main
	66	}
03513e5c	67	fetch_count.Add(1)
fa5e7c1b	68	if messages.Back() == nil \|\| !request.StartTime.Before(messages.Back().Value.(*Message).Time) {
92d1d6ac	69	waiting.PushBack(request)
03513e5c	70	fetch_wait_count.Add(1)
92d1d6ac SW	71	} else {
	72	start := messages.Back()
	73	response_size := 1
fa5e7c1b	74	if messages.Front().Value.(*Message).Time.After(request.StartTime) {
92d1d6ac SW	75	start = messages.Front()
	76	response_size = message_count
	77	} else {
fa5e7c1b	78	for start.Prev().Value.(*Message).Time.After(request.StartTime) {
92d1d6ac SW	79	start = start.Prev()
	80	response_size++
	81	}
	82	}
	83	response_messages := make([]Message, 0, response_size)
	84	for m := start; m != nil; m = m.Next() {
fa5e7c1b	85	response_messages = append(response_messages, m.Value.(Message))
92d1d6ac SW	86	}
	87	request.Messages <- response_messages
	88	}
	89	}
	90	}
	91	}
	92
	93	func start_store() Store {
fa5e7c1b	94	store := Store{make(chan Message, 20), make(chan StoreRequest, 20)}
92d1d6ac SW	95	go manage_store(store)
	96	return store
	97	}
	98
f0e385c7 SW	99	const frame_html = `<!DOCTYPE html PUBLIC "-//W3C//DTD XHTML 1.1//EN"
	100	"http://www.w3.org/TR/xhtml11/DTD/xhtml11.dtd">
	101
	102	<html xmlns="http://www.w3.org/1999/xhtml">
	103	<head>
	104	<script type="text/javascript"><!--//--><![CDATA[//><!--
	105	var since;
	106	function go() {
	107	var delay = 10000;
	108	var xhr = new XMLHttpRequest();
	109	xhr.onreadystatechange = function() {
	110	if (this.readyState == this.DONE) {
	111	if (this.status == 200) {
	112	var rtxt = this.responseText;
	113	if (rtxt != null) {
	114	var r = JSON.parse(rtxt);
	115	if (r != null) {
	116	window.parent.postMessage(rtxt, "*");
	117	delay = 40;
	118	if (r.length >= 1 && "Time" in r[r.length-1]) {
	119	since = r[r.length-1]["Time"];
	120	}
	121	}
	122	}
	123	}
	124	window.setTimeout(go, delay);
	125	}
	126	}
	127	var uri = "/fetch";
	128	if (since) {
	129	uri += '?since="' + since + '"';
	130	}
	131	xhr.open("GET", uri);
	132	xhr.send();
	133	}
	134	//--><!]]></script>
	135	</head>
	136	<body onload="go()">
	137	</body>
	138	</html>
	139	`
	140
92d1d6ac SW	141	func start_server(store Store) {
92d1d6ac SW	142	http.HandleFunc("/fetch", func(w http.ResponseWriter, r *http.Request) {
87ac1d98 SW	143	var since time.Time
	144	url_since := r.FormValue("since")
	145	if url_since != "" {
	146	err := json.Unmarshal([]byte(url_since), &since)
	147	if err != nil {
	148	log.Print("fetch: parse since: ", err)
	149	w.WriteHeader(http.StatusBadRequest)
	150	w.Write([]byte("Could not parse since as date"))
	151	return
	152	}
	153	}
92d1d6ac	154	messages_from_store := make(chan []Message, 1)
fa5e7c1b	155	store.Get <- &StoreRequest{since, messages_from_store}
92d1d6ac	156
9b33d853	157	json_encoded, err := json.Marshal(<-messages_from_store)
92d1d6ac SW	158	if err != nil {
	159	log.Print("json encode: ", err)
	160	w.WriteHeader(http.StatusInternalServerError)
	161	return
	162	}
	163	w.Header().Add("Content-Type", "application/json")
	164	w.Write(json_encoded)
	165	})
	166
	167	http.HandleFunc("/speak", func(w http.ResponseWriter, r *http.Request) {
b199796a SW	168	store.Add <- &Message{
	169	time.Now(),
	170	r.FormValue("id"),
	171	r.FormValue("text")}
92d1d6ac SW	172	})
92d1d6ac SW	173
f0e385c7	174	http.HandleFunc("/frame", func(w http.ResponseWriter, r *http.Request) {
a69e0831	175	frame_count.Add(1)
f0e385c7 SW	176	w.Write([]byte(frame_html));
	177	})
	178
86945f8d	179	log.Fatal(http.ListenAndServe(":"+strconv.Itoa(*port), nil))
92d1d6ac SW	180	}
	181
	182	func main() {
bc44b6bc	183	flag.Parse()
92d1d6ac SW	184	store := start_store()
	185	start_server(store)
	186	}