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