git.scottworley.com Git - reliable-chat/blame_incremental

... / ...

Commit	Line	Data
	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 frame_count = expvar.NewInt("frame_count")
	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
	20	type Message struct {
	21	Time time.Time
	22	ID string
	23	Text string
	24	}
	25
	26	type StoreRequest struct {
	27	StartTime time.Time
	28	Messages chan<- []Message
	29	}
	30
	31	type Store struct {
	32	Add chan *Message
	33	Get chan *StoreRequest
	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()
	43	main:
	44	for {
	45	select {
	46	case new_message, ok := <-store.Add:
	47	if !ok {
	48	break main
	49	}
	50	speak_count.Add(1)
	51	for waiter := waiting.Front(); waiter != nil; waiter = waiter.Next() {
	52	waiter.Value.(StoreRequest).Messages <- []Message{new_message}
	53	close(waiter.Value.(*StoreRequest).Messages)
	54	fetch_wake_count.Add(1)
	55	}
	56	waiting.Init()
	57	messages.PushBack(new_message)
	58	if message_count < max_messages {
	59	message_count++
	60	} else {
	61	messages.Remove(messages.Front())
	62	}
	63	case request, ok := <-store.Get:
	64	if !ok {
	65	break main
	66	}
	67	fetch_count.Add(1)
	68	if messages.Back() == nil \|\| !request.StartTime.Before(messages.Back().Value.(*Message).Time) {
	69	waiting.PushBack(request)
	70	fetch_wait_count.Add(1)
	71	} else {
	72	start := messages.Back()
	73	response_size := 1
	74	if messages.Front().Value.(*Message).Time.After(request.StartTime) {
	75	start = messages.Front()
	76	response_size = message_count
	77	} else {
	78	for start.Prev().Value.(*Message).Time.After(request.StartTime) {
	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() {
	85	response_messages = append(response_messages, m.Value.(Message))
	86	}
	87	request.Messages <- response_messages
	88	}
	89	}
	90	}
	91	}
	92
	93	func start_store() Store {
	94	store := Store{make(chan Message, 20), make(chan StoreRequest, 20)}
	95	go manage_store(store)
	96	return store
	97	}
	98
	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
	141	func start_server(store Store) {
	142	http.HandleFunc("/fetch", func(w http.ResponseWriter, r *http.Request) {
	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	}
	154	messages_from_store := make(chan []Message, 1)
	155	store.Get <- &StoreRequest{since, messages_from_store}
	156
	157	json_encoded, err := json.Marshal(<-messages_from_store)
	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) {
	168	store.Add <- &Message{
	169	time.Now(),
	170	r.FormValue("id"),
	171	r.FormValue("text")}
	172	})
	173
	174	http.HandleFunc("/frame", func(w http.ResponseWriter, r *http.Request) {
	175	frame_count.Add(1)
	176	w.Write([]byte(frame_html));
	177	})
	178
	179	log.Fatal(http.ListenAndServe(":"+strconv.Itoa(*port), nil))
	180	}
	181
	182	func main() {
	183	flag.Parse()
	184	store := start_store()
	185	start_server(store)
	186	}