.

t/neo/cmd/neostorage-go/neostorage.go

@@ -5,23 +5,47 @@
 package main
 
 import (
+	"context"
+	"flag"
+	"fmt"
+	"log"
+	"os"
+
 	//_ "../../storage"	// XXX rel ok?
 	neo "../.."
-	"fmt"
-	"context"
-	_ "time"
+
+	zodb "../../zodb"
+	_ "../../zodb/wks"
 )
 
 
 // TODO options:
 // cluster, masterv, bind ...
 
+func usage() {
+	fmt.Fprintf(os.Stderr,
+`neostorage runs one NEO storage server.
+
+Usage: neostorage [options] zstor	XXX
+`)
+}
+
 func main() {
-	// var t neo.Tid = neo.MAX_TID
-	// fmt.Printf("%T %x\n", t, t)
-	// println("TODO")
+	flag.Usage = usage
+	flag.Parse()
+	argv := flag.Args()
+
+	if len(argv) == 0 {
+		usage()
+		os.Exit(2)
+	}
+
+	zstor, err := zodb.OpenStorageURL(argv[0])
+	if err != nil {
+		log.Fatal(err)
+	}
 
-	storsrv := &neo.StorageApplication{}
+	storsrv := neo.NewStorage(zstor)
 
 	/*
 	ctx, cancel := context.WithCancel(context.Background())
@@ -32,6 +56,8 @@ func main() {
 	*/
 	ctx := context.Background()
 
-	err := neo.ListenAndServe(ctx, "tcp", "localhost:1234", storsrv)
-	fmt.Println(err)
+	err = neo.ListenAndServe(ctx, "tcp", "localhost:1234", storsrv)
+	if err != nil {
+		log.Fatal(err)
+	}
 }

t/neo/connection.go

@@ -275,6 +275,7 @@ func (nl *NodeLink) serveRecv() {
 		if handleNewConn != nil {
 			// TODO avoid spawning goroutine for each new Ask request -
 			//	- by keeping pool of read inactive goroutine / conn pool ?
+			// XXX rework interface for this to be Accept-like ?
 			go func() {
 				nl.handleWg.Add(1)
 				defer nl.handleWg.Done()
@@ -397,7 +398,7 @@ func (c *Conn) close() {
 	})
 }
 
-// Close connection
+// Close closes connection
 // Any blocked Send() or Recv() will be unblocked and return error
 //
 // NOTE for Send() - once transmission was started - it will complete in the

t/neo/storage.go

@@ -29,6 +29,10 @@ type Storage struct {
 	zstor zodb.IStorage // underlying ZODB storage	XXX temp ?
 }
 
+func NewStorage(zstor zodb.IStorage) *Storage {
+	return &Storage{zstor}
+}
+
 
 /*
 // XXX change to bytes.Buffer if we need to access it as I/O
@@ -98,13 +102,11 @@ func (stor *Storage) ServeLink(ctx context.Context, link *NodeLink) {
 }
 
 
-type StorageClientHandler struct {
-	stor *Storage
-}
+// XXX naming for RecvAndDecode and EncodeAndSend
 
 // XXX stub
 // XXX move me out of here
-func RecvAndDecode(conn *Conn) (interface{}, error) {
+func RecvAndDecode(conn *Conn) (interface{}, error) {	// XXX interface{} -> NEODecoder ?
 	pkt, err := conn.Recv()
 	if err != nil {
 		return nil, err
@@ -123,54 +125,68 @@ func EncodeAndSend(conn *Conn, pkt NEOEncoder) error {
 	return conn.Send(&buf)	// XXX why pointer?
 }
 
-// XXX naming for RecvAndDecode and EncodeAndSend
-func (ch *StorageClientHandler) ServeConn(ctx context.Context, conn *Conn) {
-	// TODO ctx.Done -> close conn
-	defer conn.Close()	// XXX err
-
-	pkt, err := RecvAndDecode(conn)
-	if err != nil {
-		return	// XXX log / err / send error before closing
-	}
-
-	switch pkt := pkt.(type) {
-	case *GetObject:
-		xid := zodb.Xid{Oid: pkt.Oid}
-		if pkt.Serial != INVALID_TID {
-			xid.Tid = pkt.Serial
-			xid.TidBefore = false
-		} else {
-			xid.Tid = pkt.Tid
-			xid.TidBefore = true
+// ServeClient serves incoming connection on which peer identified itself as client
+func (stor *Storage) ServeClient(ctx context.Context, conn *Conn) {
+	// close connection when either cancelling or returning (e.g. due to an error)
+	// ( when cancelling - conn.Close will signal to current IO to
+	//   terminate with an error )
+	retch := make(chan struct{})
+	defer func() { close(retch) }()
+	go func() {
+		select {
+		case <-ctx.Done():
+			// XXX tell client we are shutting down?
+		case <-retch:
 		}
+		conn.Close()	// XXX err
+	}()
 
-		data, tid, err := ch.stor.zstor.Load(xid)
+	for {
+		req, err := RecvAndDecode(conn)
 		if err != nil {
-			// TODO translate err to NEO protocol error codes
-			errPkt := Error{Code: 0, Message: err.Error()}
-			EncodeAndSend(conn, &errPkt)	// XXX err
-		} else {
-			answer := AnswerGetObject{
-					Oid:	 xid.Oid,
-					Serial: tid,
-
-					Compression: false,
-					Data: data,
-					// XXX .CheckSum
-
-					// XXX .NextSerial
-					// XXX .DataSerial
-				}
-			EncodeAndSend(conn, &answer)	// XXX err
+			return	// XXX log / err / send error before closing
 		}
 
-	case *LastTransaction:
-		// ----//---- for zstor.LastTid()
-
-	case *ObjectHistory:
+		switch req := req.(type) {
+		case *GetObject:
+			xid := zodb.Xid{Oid: req.Oid}
+			if req.Serial != INVALID_TID {
+				xid.Tid = req.Serial
+				xid.TidBefore = false
+			} else {
+				xid.Tid = req.Tid
+				xid.TidBefore = true
+			}
+
+			var reply NEOEncoder
+			data, tid, err := stor.zstor.Load(xid)
+			if err != nil {
+				// TODO translate err to NEO protocol error codes
+				reply = &Error{Code: 0, Message: err.Error()}
+			} else {
+				reply = &AnswerGetObject{
+						Oid:	 xid.Oid,
+						Serial: tid,
+
+						Compression: false,
+						Data: data,
+						// XXX .CheckSum
+
+						// XXX .NextSerial
+						// XXX .DataSerial
+					}
+			}
+
+			EncodeAndSend(conn, reply)	// XXX err
+
+		case *LastTransaction:
+			lastTid := stor.zstor.LastTid()
+			EncodeAndSend(conn, &AnswerLastTransaction{lastTid})	// XXX err
+
+		//case *ObjectHistory:
+		//case *StoreObject:
+		}
 
-	//case *StoreObject:
+		//req.Put(...)
 	}
-
-	//pkt.Put(...)
 }

t/neo/zodb/zodb.go

@@ -122,7 +122,7 @@ type IStorage interface {
 	// LastTid returns the id of the last committed transaction.
 	// if not transactions have been committed yet, LastTid returns Tid zero value
 	// XXX ^^^ ok ?
-	LastTid() Tid	// XXX -> Tid, ok ?
+	LastTid() Tid	// XXX -> Tid, ok ?  ,err ?
 
 	// LoadSerial and LoadBefore generalized into 1 Load  (see Xid for details)
 	// TODO data []byte -> something allocated from slab ?