diff --git a/serialization.go b/serialization.go index 72f6863..6efbb45 100644 --- a/serialization.go +++ b/serialization.go @@ -64,3 +64,21 @@ func (me *JsonSerializer) Deserialize(serialized []byte) (interface{}, error) { obj := reflect.Indirect(reflect.ValueOf(objPtr)).Interface() return obj, nil } + +type PassthruSerializer struct {} + +func NewPassthruSerializer() Serializer { + return &PassthruSerializer{} +} + +func (me PassthruSerializer) Serialize(obj interface{}) ([]byte, error) { + serialized, ok := obj.([]byte) + if !ok { + return nil, errors.New("Object is not a slice of bytes") + } + return serialized, nil +} + +func (me PassthruSerializer) Deserialize(serialized []byte) (interface{}, error) { + return serialized, nil +} diff --git a/simpleserver/simpleserver.go b/simpleserver/simpleserver.go new file mode 100644 index 0000000..20af854 --- /dev/null +++ b/simpleserver/simpleserver.go @@ -0,0 +1,97 @@ +package main + +import ( + "fmt" + "github.com/pebbe/zmq4" + "github.com/satori/go.uuid" + "bitbucket.org/nicdex/adaptech-goes" + "os" + "path" + "encoding/binary" +) + +func main() { + fmt.Println("Simple ZeroMQ server for goes.") + + //TODO: config/flag + storagePath := path.Join(os.TempDir(), uuid.NewV4().String()) + + goes.SetStorage(goes.NewDiskStorage(storagePath)) + goes.SetSerializer(goes.NewPassthruSerializer()) + + context, err := zmq4.NewContext() + if err != nil { + panic(err) + } + defer context.Term() + + replySocket, err := context.NewSocket(zmq4.REP) + if err != nil { + panic(err) + } + defer replySocket.Close() + + //TODO: config/flag + listenAddr := "tcp://*:12345" + err = replySocket.Bind(listenAddr) + if err != nil { + panic(err) + } + + for { + message, err := replySocket.RecvMessageBytes(0) + if err != nil { + fmt.Println("Error receiving command from client", err) + continue + } + + command := string(message[0]) + switch command { + case "AddEvent": + aggregateId, err := uuid.FromBytes(message[1]) + if err != nil { + fmt.Println("Wrong format for AggregateId", err) + break + } + fmt.Println(command, aggregateId.String()) + data := message[2] + err = goes.AddEvent(goes.Event{aggregateId, data}) + if err != nil { + panic(err) + } + replySocket.Send("Ok", 0) + case "ReadStream": + aggregateId, err := uuid.FromBytes(message[1]) + if err != nil { + fmt.Println("Wrong format for AggregateId", err) + break + } + fmt.Println(command, aggregateId.String()) + events, err := goes.RetrieveFor(aggregateId) + if err != nil { + panic(err) + } + sendEvents(replySocket, events) + case "ReadAll": + fmt.Println(command) + events, err := goes.RetrieveAll() + if err != nil { + panic(err) + } + sendEvents(replySocket, events) + } + } +} + +func sendEvents(socket *zmq4.Socket, events []*goes.Event) { + len := len(events) + lenBytes := make([]byte, 8) + binary.BigEndian.PutUint64(lenBytes, uint64(len)) + socket.SendBytes(lenBytes, zmq4.SNDMORE) + + i := 0 + for ; i < len-1; i++ { + socket.SendBytes(events[i].Payload.([]byte), zmq4.SNDMORE) + } + socket.SendBytes(events[i].Payload.([]byte), 0) +} \ No newline at end of file