msgid, err id structs; copy buf to prevent overwriting.
@@ -7,16 +7,32 @@ import (
// Chan is a delimited duplex channel. It is used to have a channel interface
// around a delimited.Reader or Writer.
type Chan struct {
- MsgChan chan []byte
- ErrChan chan error
+ MsgChan chan MsgAndID
+ ErrChan chan ErrAndID
CloseChan chan bool
}
+// MsgAndID is a struct for passing a message paired with an id,
+// we need the id to report errors with this message id later
+// ID is nil on an incoming message
+type MsgAndID struct {
+ ID []byte
+ Msg []byte
+}
+
+// ErrAndID is a struct that holds an error and an id
+// it is used to report errors regarding the specific message id.
+// ID is nil when the err is incoming error message (we don't know the id)
+type ErrAndID struct {
+ ID []byte
+ Err error
+}
+
// NewChan constructs a Chan with a given buffer size.
func NewChan(chanSize int) *Chan {
return &Chan{
- MsgChan: make(chan []byte, chanSize),
- ErrChan: make(chan error, 1),
+ MsgChan: make(chan MsgAndID, chanSize),
+ ErrChan: make(chan ErrAndID, 1),
CloseChan: make(chan bool, 2),
}
}
@@ -38,17 +54,21 @@ Loop:
if err == io.EOF {
break Loop // done
}
-
// unexpected error. tell the client.
- s.ErrChan <- err
+ s.ErrChan <- ErrAndID{nil, err}
break Loop
}
select {
case <-s.CloseChan:
break Loop // told we're done
- case s.MsgChan <- buf:
+ default:
+ if buf != nil {
+ var emptybuf []byte
+ copy(emptybuf, buf)
// ok seems fine. send it away
+ s.MsgChan <- MsgAndID{nil, emptybuf}
+ }
}
}
@@ -76,10 +96,10 @@ Loop:
break Loop
}
- if _, err := mw.WriteRecord(msg); err != nil {
+ if _, err := mw.WriteRecord(msg.Msg); err != nil {
if err != io.EOF {
// unexpected error. tell the client.
- s.ErrChan <- err
+ s.ErrChan <- ErrAndID{msg.ID, err}
}
break Loop
