scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-10-10 21:07:53 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Refactor

Browse source
This commit is contained in:
Cody Soyland 2014-02-17 16:59:47 -06:00
parent 146dc92acc
commit 4d23b5af03
1 changed files with 27 additions and 53 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

80
core/http.go
View file

@ -282,19 +282,7 @@ func (self *WebService) HandleListen(w http.ResponseWriter, r *http.Request) {
`))
}
func (self *WebService) HandleListenWS(w http.ResponseWriter, r *http.Request) {
defer func() {
err := recover()
spew.Dump(err)
}()
ws, err := websocket.Upgrade(w, r, nil, 1024, 1024)
if _, ok := err.(websocket.HandshakeError); ok {
http.Error(w, "Not a websocket handshake", 400)
return
} else if err != nil {
log.Println(err)
return
}
func (self *WebService) streamer(writer func(map[string]interface{}) error) {
inbox := make(chan interface{}, 10)
outbox := make(chan interface{}, 10)
notify.Start("inbox", inbox)
@ -318,7 +306,7 @@ func (self *WebService) HandleListenWS(w http.ResponseWriter, r *http.Request) {
typ := reflect.TypeOf(data)
err := ws.WriteJSON(map[string]interface{}{
err := writer(map[string]interface{}{
"type": typ.String(),
"dump": spew.Sdump(data),
"host": host,
@ -334,6 +322,24 @@ func (self *WebService) HandleListenWS(w http.ResponseWriter, r *http.Request) {
}
}
func (self *WebService) HandleListenWS(w http.ResponseWriter, r *http.Request) {
defer func() {
err := recover()
spew.Dump(err)
}()
ws, err := websocket.Upgrade(w, r, nil, 1024, 1024)
if _, ok := err.(websocket.HandshakeError); ok {
http.Error(w, "Not a websocket handshake", 400)
return
} else if err != nil {
log.Println(err)
return
}
self.streamer(func(data map[string]interface{}) error {
return ws.WriteJSON(data)
})
}
func drain(ch chan interface{}) {
for {
switch {
@ -349,43 +355,11 @@ func (self *WebService) HandleListenStream(w http.ResponseWriter, r *http.Reques
err := recover()
spew.Dump(err)
}()
inbox := make(chan interface{}, 10)
outbox := make(chan interface{}, 10)
notify.Start("inbox", inbox)
notify.Start("outbox", outbox)
var obj interface{}
var data interface{}
var inmessage *db.Message
var outmessage *db.Envelope
var host string
for {
select {
case obj = <-outbox:
outmessage = obj.(*db.Envelope)
data = outmessage.Message.Data
host = outmessage.Host.String()
case obj = <-inbox:
inmessage = obj.(*db.Message)
data = inmessage.Data
host = ""
}
typ := reflect.TypeOf(data)
writer := json.NewEncoder(w)
err := writer.Encode(map[string]interface{}{
"type": typ.String(),
"dump": spew.Sdump(data),
"host": host,
})
w.(http.Flusher).Flush()
if err != nil {
log.Println("stopping")
notify.Stop("inbox", inbox)
notify.Stop("outbox", outbox)
drain(inbox)
drain(outbox)
return
}
}
writer := json.NewEncoder(w)
flusher := w.(http.Flusher)
self.streamer(func(data map[string]interface{}) error {
err := writer.Encode(data)
flusher.Flush()
return err
})
}