Commit a41e0b4d authored by Kirill Smelkov's avatar Kirill Smelkov

.

parent 7025320e
...@@ -372,7 +372,7 @@ func (stor *Storage) ServeLink(ctx context.Context, link *neo.NodeLink) { ...@@ -372,7 +372,7 @@ func (stor *Storage) ServeLink(ctx context.Context, link *neo.NodeLink) {
var serveConn func(context.Context, *neo.Conn) var serveConn func(context.Context, *neo.Conn)
switch nodeInfo.NodeType { switch nodeInfo.NodeType {
case neo.CLIENT: case neo.CLIENT:
serveConn = stor.ServeClient serveConn = stor.serveClient
default: default:
// XXX vvv should be reply to peer // XXX vvv should be reply to peer
...@@ -388,7 +388,7 @@ func (stor *Storage) ServeLink(ctx context.Context, link *neo.NodeLink) { ...@@ -388,7 +388,7 @@ func (stor *Storage) ServeLink(ctx context.Context, link *neo.NodeLink) {
break break
} }
// XXX wrap conn close to happen here, not in ServeClient ? // XXX wrap conn close to happen here, not in serveClient ?
go serveConn(ctx, conn) go serveConn(ctx, conn)
} }
...@@ -396,9 +396,9 @@ func (stor *Storage) ServeLink(ctx context.Context, link *neo.NodeLink) { ...@@ -396,9 +396,9 @@ func (stor *Storage) ServeLink(ctx context.Context, link *neo.NodeLink) {
} }
// ServeClient serves incoming connection on which peer identified itself as client // serveClient serves incoming connection on which peer identified itself as client
// XXX +error return? // XXX +error return?
func (stor *Storage) ServeClient(ctx context.Context, conn *neo.Conn) { func (stor *Storage) serveClient(ctx context.Context, conn *neo.Conn) {
fmt.Printf("stor: %s: serving new client conn\n", conn) fmt.Printf("stor: %s: serving new client conn\n", conn)
// rederive ctx to be also cancelled if M tells us StopOperation // rederive ctx to be also cancelled if M tells us StopOperation
...@@ -418,62 +418,67 @@ func (stor *Storage) ServeClient(ctx context.Context, conn *neo.Conn) { ...@@ -418,62 +418,67 @@ func (stor *Storage) ServeClient(ctx context.Context, conn *neo.Conn) {
}() }()
for { for {
req, err := conn.Recv() serveClient1(conn) // XXX err check
if err != nil { }
return // XXX log / err / send error before closing }
}
switch req := req.(type) { // serveClient1 serves 1 request from a client
case *neo.GetObject: func (stor *Storage) serveClient1(conn *neo.Conn) error {
xid := zodb.Xid{Oid: req.Oid} req, err := conn.Recv()
if req.Serial != neo.INVALID_TID { if err != nil {
xid.Tid = req.Serial return // XXX log / err / send error before closing
xid.TidBefore = false }
} else {
xid.Tid = req.Tid
xid.TidBefore = true
}
var reply neo.Msg switch req := req.(type) {
data, tid, err := stor.zstor.Load(xid) case *neo.GetObject:
if err != nil { xid := zodb.Xid{Oid: req.Oid}
// TODO translate err to NEO protocol error codes if req.Serial != neo.INVALID_TID {
reply = neo.ErrEncode(err) xid.Tid = req.Serial
} else { xid.TidBefore = false
reply = &neo.AnswerGetObject{ } else {
Oid: xid.Oid, xid.Tid = req.Tid
Serial: tid, xid.TidBefore = true
}
Compression: false,
Data: data,
// XXX .CheckSum
// XXX .NextSerial
// XXX .DataSerial
}
}
conn.Send(reply) // XXX err var reply neo.Msg
data, tid, err := stor.zstor.Load(xid)
if err != nil {
// TODO translate err to NEO protocol error codes
reply = neo.ErrEncode(err)
} else {
reply = &neo.AnswerGetObject{
Oid: xid.Oid,
Serial: tid,
Compression: false,
Data: data,
// XXX .CheckSum
// XXX .NextSerial
// XXX .DataSerial
}
}
case *neo.LastTransaction: conn.Send(reply) // XXX err
var reply neo.Msg
lastTid, err := stor.zstor.LastTid() case *neo.LastTransaction:
if err != nil { var reply neo.Msg
reply = neo.ErrEncode(err)
} else {
reply = &neo.AnswerLastTransaction{lastTid}
}
conn.Send(reply) // XXX err lastTid, err := stor.zstor.LastTid()
if err != nil {
reply = neo.ErrEncode(err)
} else {
reply = &neo.AnswerLastTransaction{lastTid}
}
//case *ObjectHistory: conn.Send(reply) // XXX err
//case *StoreObject:
default: //case *ObjectHistory:
panic("unexpected packet") // XXX //case *StoreObject:
}
//req.Put(...) default:
panic("unexpected packet") // XXX
} }
//req.Put(...)
} }
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment