scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-09-06 00:25:55 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Add websocket listener for messaging layer.

Browse source
This commit is contained in:
Cody Soyland 2014-02-11 18:28:55 -06:00
parent c258de14a8
commit d4dd91973c
4 changed files with 118 additions and 38 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

104
core/http.go
View file

@ -7,10 +7,13 @@ import (
"net/http"
"pilosa/config"
"pilosa/db"
"reflect"
"runtime"
"strconv"
notify "github.com/bitly/go-notify"
"github.com/davecgh/go-spew/spew"
"github.com/gorilla/websocket"
"tux21b.org/v1/gocql/uuid"
)
@ -31,6 +34,7 @@ func (self *WebService) Run() {
mux.HandleFunc("/stats", self.HandleStats)
mux.HandleFunc("/info", self.HandleInfo)
mux.HandleFunc("/processes", self.HandleProcesses)
mux.HandleFunc("/listen/ws", self.HandleListenWS)
mux.HandleFunc("/listen", self.HandleListen)
mux.HandleFunc("/test", self.HandleTest)
mux.HandleFunc("/version", self.HandleVersion)
@ -249,16 +253,92 @@ func (self *WebService) HandlePing(w http.ResponseWriter, r *http.Request) {
}
func (self *WebService) HandleListen(w http.ResponseWriter, r *http.Request) {
//listener := service.NewListener()
//encoder := json.NewEncoder(w)
//for {
// select {
// case message := <-listener:
// err := encoder.Encode(message)
// if err != nil {
// log.Println("Error sending message")
// return
// }
// }
//}
w.Write([]byte(`
<html><head><title>Pilosa - streaming client</title></head><body>
<style type="text/css">
dt.inbox { background-color: #efe; }
dt.outbox { background-color: #eef; }
</style>
<script src="//ajax.googleapis.com/ajax/libs/jquery/1.10.2/jquery.min.js"></script>
<script type="text/javascript">
$dl = $("<dl/>")
$dl.on('click', 'dt', function(e) {
$(this).next().toggle()
})
$("body").append($dl)
ws = new WebSocket("ws://" + window.location.host + "/listen/ws")
ws.onmessage = function(mes) {
obj = JSON.parse(mes.data)
if (obj.host) {
$dl.append('<dt class="outbox">&rarr;' + obj.type + ' (to: ' + obj.host + ')</dt>')
} else {
$dl.append('<dt class="inbox">&larr;' + obj.type + '</dt>')
}
$dl.append('<dd style="display:none;"><pre>' + obj.dump + obj.host + '</pre></dd>')
}
</script>
</body></html>
`))
}
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
}
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)
err := ws.WriteJSON(map[string]interface{}{
"type": typ.String(),
"dump": spew.Sdump(data),
"host": host,
})
if err != nil {
log.Println("stopping")
notify.Stop("inbox", inbox)
notify.Stop("outbox", outbox)
drain(inbox)
drain(outbox)
return
}
}
}
func drain(ch chan interface{}) {
for {
switch {
case <-ch:
default:
return
}
}
}

16
db/db.go
View file

@ -10,21 +10,11 @@ type Message struct {
Data interface{} `json:data`
}
/*
import "pilosa/core"
type Message interface {
Handle(*core.Service)
type Envelope struct {
Message *Message
Host *uuid.UUID
}
type Message struct {
Key string `json:key`
Data interface{} `json:data`
Destination Location
}
*/
type HoldResult interface {
ResultId() *uuid.UUID
ResultData() interface{}

10
deps.json
View file

@ -63,5 +63,15 @@
"repo": "github.com/davecgh/go-spew/spew",
"version": "e762b3d1320b76030bd7f6cc2bfc3d9acce874c0",
"type": "git"
},
"notify": {
"repo": "github.com/bitly/go-notify",
"version": "0a148b8111d688ba7550fc7119fe0d5d8e650838",
"type": "git"
},
"websocket": {
"repo": "github.com/gorilla/websocket",
"version": "92334662baa9cbebc2e6e68b8d56bc1233f85a4c",
"type": "git"
}
}

26
transport/tcp.go
View file

@ -10,14 +10,10 @@ import (
"pilosa/db"
"time"
notify "github.com/bitly/go-notify"
"tux21b.org/v1/gocql/uuid"
)
type envelope struct {
message *db.Message
host *uuid.UUID
}
type connection struct {
transport *TcpTransport
inbox chan *db.Message
@ -106,7 +102,7 @@ type TcpTransport struct {
service *core.Service
port int
inbox chan *db.Message
outbox chan envelope
outbox chan db.Envelope
connections map[uuid.UUID]*connection
reg chan *newconnection
}
@ -117,13 +113,13 @@ func (self *TcpTransport) Run() {
for {
select {
case env := <-self.outbox:
con, ok := self.connections[*(env.host)]
con, ok := self.connections[*(env.Host)]
if !ok {
con = &connection{self, make(chan *db.Message, 100), make(chan *db.Message, 100), nil, env.host}
con = &connection{self, make(chan *db.Message, 100), make(chan *db.Message, 100), nil, env.Host}
go con.manage()
self.connections[*env.host] = con
self.connections[*env.Host] = con
}
con.outbox <- env.message
con.outbox <- env.Message
case nc := <-self.reg:
self.connections[*nc.id] = nc.connection
}
@ -157,11 +153,15 @@ func (self *TcpTransport) Close() {
}
func (self *TcpTransport) Send(message *db.Message, host *uuid.UUID) {
self.outbox <- envelope{message, host}
envelope := db.Envelope{message, host}
notify.Post("outbox", &envelope)
self.outbox <- envelope
}
func (self *TcpTransport) Receive() *db.Message {
return <-self.inbox
message := <-self.inbox
notify.Post("inbox", message)
return message
}
func (self *TcpTransport) Push(message *db.Message) {
@ -169,5 +169,5 @@ func (self *TcpTransport) Push(message *db.Message) {
}
func NewTcpTransport(service *core.Service) *TcpTransport {
return &TcpTransport{service, config.GetInt("port_tcp"), make(chan *db.Message, 100), make(chan envelope, 100), make(map[uuid.UUID]*connection), make(chan *newconnection)}
return &TcpTransport{service, config.GetInt("port_tcp"), make(chan *db.Message, 100), make(chan db.Envelope, 100), make(map[uuid.UUID]*connection), make(chan *newconnection)}
}