Fix the rest of the code

macressler · Dec 23, 2015 · 536216f · 536216f
1 parent 02aca81
commit 536216f
Show file tree

Hide file tree

Showing 2 changed files with 9 additions and 6 deletions.
diff --git a/examples/pubsub/main.go b/examples/pubsub/main.go
@@ -33,8 +33,9 @@ func pub() {
 }
 
 func sub() {
-	_, err := broker.Subscribe(topic, func(msg *broker.Message) {
-		fmt.Println("[sub] received message:", string(msg.Body), "header", msg.Header)
+	_, err := broker.Subscribe(topic, func(p broker.Publication) error {
+		fmt.Println("[sub] received message:", string(p.Message().Body), "header", p.Message().Header)
+		return nil
 	})
 	if err != nil {
 		fmt.Println(err)

diff --git a/server/subscriber.go b/server/subscriber.go
@@ -155,11 +155,12 @@ func validateSubscriber(sub Subscriber) error {
 }
 
 func (s *rpcServer) createSubHandler(sb *subscriber, opts options) broker.Handler {
-	return func(msg *broker.Message) {
+	return func(p broker.Publication) error {
+		msg := p.Message()
 		ct := msg.Header["Content-Type"]
 		cf, err := s.newCodec(ct)
 		if err != nil {
-			return
+			return err
 		}
 
 		hdr := make(map[string]string)
@@ -190,11 +191,11 @@ func (s *rpcServer) createSubHandler(sb *subscriber, opts options) broker.Handle
 			defer co.Close()
 
 			if err := co.ReadHeader(&codec.Message{}, codec.Publication); err != nil {
-				continue
+				return err
 			}
 
 			if err := co.ReadBody(req.Interface()); err != nil {
-				continue
+				return err
 			}
 
 			fn := func(ctx context.Context, msg Publication) error {
@@ -225,6 +226,7 @@ func (s *rpcServer) createSubHandler(sb *subscriber, opts options) broker.Handle
 				message:     req.Interface(),
 			})
 		}
+		return nil
 	}
 }