Merge pull request #609 from ripienaar/606

ensure response message goes to the reply subject in RespondMsg

nats.go

@@ -3917,6 +3917,7 @@ func (m *Msg) RespondMsg(msg *Msg) error {
 	if m.Reply == "" {
 		return ErrMsgNoReply
 	}
+	msg.Subject = m.Reply
 	m.Sub.mu.Lock()
 	nc := m.Sub.conn
 	m.Sub.mu.Unlock()

nats_test.go

@@ -2646,3 +2646,40 @@ func TestJSSubscribe(t *testing.T) {
 		t.Fatalf("subscriber not subscribed to the delivery subject")
 	}
 }
+
+func TestMsg_RespondMsg(t *testing.T) {
+	s := RunServerOnPort(-1)
+	defer s.Shutdown()
+
+	nc, err := Connect(s.ClientURL())
+	if err != nil {
+		t.Fatalf("Expected to connect to server, got %v", err)
+	}
+	defer nc.Close()
+
+	sub, err := nc.SubscribeSync(NewInbox())
+	if err != nil {
+		t.Fatalf("subscribe failed: %s", err)
+	}
+
+	nc.PublishMsg(&Msg{Reply: sub.Subject, Subject: sub.Subject, Data: []byte("request")})
+	req, err := sub.NextMsg(time.Second)
+	if err != nil {
+		t.Fatalf("NextMsg failed: %s", err)
+	}
+
+	// verifies that RespondMsg sets the reply subject on msg based on req
+	err = req.RespondMsg(&Msg{Data: []byte("response")})
+	if err != nil {
+		t.Fatalf("RespondMsg failed: %s", err)
+	}
+
+	resp, err := sub.NextMsg(time.Second)
+	if err != nil {
+		t.Fatalf("NextMsg failed: %s", err)
+	}
+
+	if !bytes.Equal(resp.Data, []byte("response")) {
+		t.Fatalf("did not get correct response: %q", resp.Data)
+	}
+}