upgrade broker.Message interface (#233)

Author: dantedenis

broker/broker.go

@@ -78,6 +78,8 @@ type Message interface {
 	Unmarshal(dst any, opts ...codec.Option) error
 	// Ack acknowledge message if supported.
 	Ack() error
+
+	Error() error
 }
 
 // Subscriber is a convenience return type for the Subscribe method

broker/memory/memory.go

@@ -33,6 +33,7 @@ type memoryMessage struct {
 	body  []byte
 	hdr   metadata.Metadata
 	opts  broker.MessageOptions
+	err   error
 }
 
 func (m *memoryMessage) Ack() error {
@@ -59,6 +60,10 @@ func (m *memoryMessage) Unmarshal(dst interface{}, opts ...codec.Option) error {
 	return m.c.Unmarshal(m.body, dst)
 }
 
+func (m *memoryMessage) Error() error {
+	return m.err
+}
+
 type Subscriber struct {
 	ctx     context.Context
 	exit    chan bool

broker/noop.go

@@ -100,6 +100,7 @@ type noopMessage struct {
 	body []byte
 	hdr  metadata.Metadata
 	opts MessageOptions
+	err  error
 }
 
 func (m *noopMessage) Ack() error {
@@ -126,6 +127,10 @@ func (m *noopMessage) Unmarshal(dst interface{}, opts ...codec.Option) error {
 	return m.c.Unmarshal(m.body, dst)
 }
 
+func (m *noopMessage) Error() error {
+	return m.err
+}
+
 func (b *NoopBroker) NewMessage(ctx context.Context, hdr metadata.Metadata, body interface{}, opts ...MessageOption) (Message, error) {
 	options := NewMessageOptions(opts...)
 	if options.ContentType == "" {