refactor ginormous read loop
@@ -42,10 +42,13 @@ func (p *Protocol) readLoop() {
// Channel is closed.
break
}
-
//todo add buffer and option to limit number of concurrent goroutines
+ go p.handleMessage(msg)
+ }
+
+}
- go func(msg service.Message) {
+func (p *Protocol) handleMessage(msg service.Message) {
headers := &pb.MessageWrapper{}
if err := proto.Unmarshal(msg.Data(), headers); err != nil {
@@ -54,19 +57,30 @@ func (p *Protocol) readLoop() {
}
if headers.Req {
+ p.handleRequestMessage(msg.Sender().String(), headers)
+ } else {
+ p.handleResponseMessage(headers)
+ }
+
+}
+
+func (p *Protocol) handleRequestMessage(sender string, headers *pb.MessageWrapper) {
if payload := p.msgRequestHandlers[string(headers.Type)](headers.Payload); payload != nil {
rmsg, fParseErr := proto.Marshal(&pb.MessageWrapper{Req: false, ReqID: headers.ReqID, Type: headers.Type, Payload: payload})
if fParseErr != nil {
log.Error("Error Parsing Protocol message, err:", fParseErr)
return
}
- sendErr := p.network.SendMessage(msg.Sender().PublicKey().String(), p.name, rmsg)
+ sendErr := p.network.SendMessage(sender, p.name, rmsg)
if sendErr != nil {
log.Error("Error sending response message, err:", sendErr)
}
return
}
- } else {
+
+}
+
+func (p *Protocol) handleResponseMessage(headers *pb.MessageWrapper) {
reqId, err := uuid.FromBytes(headers.ReqID)
if err != nil {
log.Error("Error Parsing message request id, err:", err)
@@ -86,10 +100,6 @@ func (p *Protocol) readLoop() {
}
}
}
- }(msg)
- }
-
-}
func (p *Protocol) RegisterMsgHandler(msgType string, reqHandler func(msg []byte) []byte) {
p.msgRequestHandlers[msgType] = reqHandler
