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

Merge branch 'master' of ops:nuevo-pilosa

Browse source
This commit is contained in:
Todd Gruben 2014-01-17 19:19:07 -06:00
commit ccd36e351e
2 changed files with 110 additions and 34 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
hold/hold_test.go
View file

@ -22,9 +22,9 @@ func TestHoldChan(t *testing.T) {
Convey("get then set", t, func() {
id := uuid.RandomUUID()
go func() {
time.Sleep(time.Second / 10)
Hold.Set(&id, "derpsy", 10)
}()
time.Sleep(time.Second / 10)
derp, _ := Hold.Get(&id, 10)
So(derp, ShouldEqual, "derpsy")
})

142
transport/tcp.go
View file

@ -9,20 +9,106 @@ import (
"pilosa/core"
"pilosa/db"
"time"
"tux21b.org/v1/gocql/uuid"
)
type qmessage struct {
type envelope struct {
message *db.Message
host *uuid.UUID
}
type connection struct {
transport *TcpTransport
inbox chan *db.Message
outbox chan *db.Message
conn *net.Conn
process *uuid.UUID
}
type newconnection struct {
id *uuid.UUID
connection *connection
}
func init() {
gob.Register(uuid.UUID{})
}
func (self *connection) manage() {
BeginManageConnection:
for {
log.Println("manage", self)
if self.conn == nil {
process, err := self.transport.service.ProcessMap.GetProcess(self.process)
if err != nil {
log.Println("transport/tcp: error getting process, retrying in 2 seconds... ", self.process, err)
time.Sleep(2 * time.Second)
continue
}
host_string := fmt.Sprintf("%s:%d", process.Host(), process.PortTcp())
conn, err := net.Dial("tcp", host_string)
if err != nil {
log.Println("transport/tcp: error dialing: ", host_string, " Retrying in 2 seconds...")
time.Sleep(2 * time.Second)
continue
}
self.conn = &conn
go func() {
self.outbox <- &db.Message{self.transport.service.Id}
}()
}
encoder := gob.NewEncoder(*self.conn)
decoder := gob.NewDecoder(*self.conn)
var exit = make(chan int)
go func() {
for {
var mess *db.Message
err := decoder.Decode(&mess)
if err != nil {
log.Println("transport/tcp: error decoding message: ", err.Error())
exit <- 1
return
}
self.inbox <- mess
}
}()
for {
select {
case message := <-self.outbox:
err := encoder.Encode(message)
if err != nil {
log.Println(err.Error())
return
}
case message := <-self.inbox:
identifier, ok := message.Data.(uuid.UUID)
if ok {
// message is connection registration; bypass inbox and register
self.process = &identifier
self.transport.reg <- &newconnection{&identifier, self}
} else {
self.transport.inbox <- message
}
case <-exit:
if self.process != nil {
self.conn = nil
continue BeginManageConnection
} else {
return
}
}
}
}
}
type TcpTransport struct {
service *core.Service
port int
inbox chan *db.Message
outbox chan qmessage
stop chan bool
service *core.Service
port int
inbox chan *db.Message
outbox chan envelope
connections map[uuid.UUID]*connection
reg chan *newconnection
}
func (self *TcpTransport) Run() {
@ -30,22 +116,16 @@ func (self *TcpTransport) Run() {
go self.listen()
for {
select {
case qmessage := <-self.outbox:
// todo: persistent connections
process, err := self.service.ProcessMap.GetProcess(qmessage.host)
if err != nil {
log.Println("transport/tcp", err)
return
case env := <-self.outbox:
con, ok := self.connections[*(env.host)]
if !ok {
con = &connection{self, make(chan *db.Message, 100), make(chan *db.Message, 100), nil, env.host}
go con.manage()
self.connections[*env.host] = con
}
host_string := fmt.Sprintf("%s:%d", process.Host(), process.PortTcp())
conn, err := net.Dial("tcp", host_string)
encoder := gob.NewEncoder(conn)
err = encoder.Encode(qmessage.message)
if err != nil {
log.Println(err.Error())
return
}
//time.Sleep(1 * time.Second)
con.outbox <- env.message
case nc := <-self.reg:
self.connections[*nc.id] = nc.connection
}
}
}
@ -54,26 +134,22 @@ func (self *TcpTransport) listen() {
port_string := fmt.Sprintf(":%d", self.port)
l, e := net.Listen("tcp", port_string)
if e != nil {
log.Fatal("Cannot bind to port!", self.port)
log.Fatal("Cannot bind to port! ", self.port)
}
for {
conn, err := l.Accept()
if err != nil {
log.Fatal("Cannot accept message!")
log.Println("Error accepting, trying again in 2 sec... ", err)
time.Sleep(2 * time.Second)
continue
}
go self.manage(&conn)
}
}
func (self *TcpTransport) manage(conn *net.Conn) {
decoder := gob.NewDecoder(*conn)
var mess *db.Message
err := decoder.Decode(&mess)
if err != nil {
log.Println("tcp/transport", err.Error())
return
}
self.inbox <- mess
con := &connection{self, make(chan *db.Message, 100), make(chan *db.Message, 100), conn, nil}
con.manage()
}
func (self *TcpTransport) Close() {
@ -81,7 +157,7 @@ func (self *TcpTransport) Close() {
}
func (self *TcpTransport) Send(message *db.Message, host *uuid.UUID) {
self.outbox <- qmessage{message, host}
self.outbox <- envelope{message, host}
}
func (self *TcpTransport) Receive() *db.Message {
@ -93,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), make(chan qmessage), nil}
return &TcpTransport{service, config.GetInt("port_tcp"), make(chan *db.Message, 100), make(chan envelope, 100), make(map[uuid.UUID]*connection), make(chan *newconnection)}
}