Repository: nats-io/nats-server
Pull request #1729 (2020-11-20 18:18:48)

Enable JetStream streams and consumer access to be exported
This allows basic access to publish into a stream and receive the pub ack across accounts, as well as pull and push based receipt of consumer messages and the ability to ack those messages.

Signed-off-by: Derek Collison <derek@nats.io>

/cc @nats-io/core



Patch:
diff --git a/server/accounts.go b/server/accounts.go
index 669067bd..82c35ed0 100644
--- a/server/accounts.go
+++ b/server/accounts.go
@@ -1631,6 +1631,7 @@ func (a *Account) addServiceImport(dest *Account, from, to string, claim *jwt.Im
 		if to == from {
 			usePub = true
 		} else {
+			from, _ = transformUntokenize(from)
 			// Create a transform
 			if tr, err = newTransform(from, transformTokenize(to)); err != nil {
 				a.mu.Unlock()
diff --git a/server/client.go b/server/client.go
index 71b57841..048e4f1e 100644
--- a/server/client.go
+++ b/server/client.go
@@ -2949,14 +2949,13 @@ func (c *client) deliverMsg(sub *subscription, subject, reply, mh, msg []byte, g
 	client.outBytes += msgSize
 
 	// Check for internal subscriptions.
-	if sub.icb != nil || client.kind == SYSTEM || client.kind == JETSTREAM || client.kind == ACCOUNT {
+	if sub.icb != nil {
 		client.mu.Unlock()
-		// Internal account clients are for service imports and need the
-		// complete raw msg with '\r\n'.
+		// Internal account clients are for service imports and need the '\r\n'.
 		if client.kind == ACCOUNT {
-			sub.icb(sub, c, string(subject), string(c.pa.reply), msg)
+			sub.icb(sub, c, string(subject), string(reply), msg)
 		} else {
-			sub.icb(sub, c, string(subject), string(c.pa.reply), msg[:msgSize])
+			sub.icb(sub, c, string(subject), string(reply), msg[:msgSize])
 		}
 		return true
 	}
@@ -3378,6 +3377,7 @@ func (c *client) processInboundClientMsg(msg []byte) bool {
 			atomic.LoadInt64(&c.srv.gateway.totalQSubs) > 0 {
 			flag |= pmrCollectQueueNames
 		}
+
 		didDeliver, qnames = c.processMsgResults(c.acc, r, msg, c.pa.deliver, c.pa.subject, c.pa.reply, flag)
 	}
 
@@ -3501,8 +3501,7 @@ func (c *client) processServiceImport(si *serviceImport, acc *Account, msg []byt
 	// TODO(dlc) - restrict to configured service imports and not responses?
 	tracking, headers := shouldSample(si.latency, c)
 	if len(c.pa.reply) > 0 {
-		rsi = c.setupResponseServiceImport(acc, si, tracking, headers)
-		if rsi != nil {
+		if rsi = c.setupResponseServiceImport(acc, si, tracking, headers); rsi != nil {
 			nrr = []byte(rsi.from)
 		}
 	} else {
@@ -3531,7 +3530,7 @@ func (c *client) processServiceImport(si *serviceImport, acc *Account, msg []byt
 	}
 
 	// Now check to see if this account has mappings that could affect the service import.
-	// Can't use non locked trick like in processInboundClientMsg, so just call into selectMappedSubject
+	// Can't use non-locked trick like in processInboundClientMsg, so just call into selectMappedSubject
 	// so we only lock once.
 	to, _ = si.acc.selectMappedSubject(to)
 
@@ -3706,8 +3705,9 @@ func (c *client) processMsgResults(acc *Account, r *SublistResult, msg, deliver,
 			}
 			continue
 		}
-		// Assume delivery subject is normal subject to this point.
+		// Assume delivery subject is the normal subject to this point.
 		dsubj = subj
+
 		// Check for stream import mapped subs (shadow subs). These apply to local subs only.
 		if sub.im != nil {
 			// If this message was a service import do not re-export to an exported stream.
@@ -3715,14 +3715,26 @@ func (c *client) processMsgResults(acc *Account, r *SublistResult, msg, deliver,
 				continue
 			}
 			if sub.im.tr != nil {
-				to, _ := sub.im.tr.transformSubject(string(subj))
+				to, _ := sub.im.tr.transformSubject(string(dsubj))
 				dsubj = append(_dsubj[:0], to...)
 			} else if sub.im.usePub {
 				dsubj = append(_dsubj[:0], subj...)
 			} else {
 				dsubj = append(_dsubj[:0], sub.im.to...)
 			}
+			// If we are mapping for a deliver subject we will reverse roles.
+			// The original subj we set from above is correct for the msg header,
+			// but we need to transform the deliver subject to properly route.
+			if len(deliver) > 0 {
+				dsubj, subj = subj, dsubj
+			}
 		}
+
+		// Remap to original if internal.
+		if sub.icb != nil {
+			subj = subject
+		}
+
 		// Normal delivery
 		mh := c.msgHeader(dsubj, creply, sub)
 		didDeliver = c.deliverMsg(sub, subj, creply, mh, msg, rplyHasGWPrefix) || didDeliver
diff --git a/server/events.go b/server/events.go
index 4d5b7330..449198cd 100644
--- a/server/events.go
+++ b/server/events.go
@@ -305,12 +305,14 @@ RESET:
 				}
 			}
 			c.mu.Lock()
+
 			// We can have an override for account here.
 			if pm.acc != nil {
 				c.acc = pm.acc
 			} else {
 				c.acc = sysacc
 			}
+
 			// Prep internal structures needed to send message.
 			c.pa.subject = []byte(pm.sub)
 			c.pa.size = len(b)
@@ -1618,7 +1620,7 @@ func (s *Server) debugSubscribers(sub *subscription, c *client, subject, reply s
 
 	// If we are only local, go ahead and return.
 	if expected == 0 {
-		s.sendInternalAccountMsg(c.acc, reply, nsubs)
+		s.sendInternalAccountMsg(nil, reply, nsubs)
 		return
 	}
 
@@ -1662,7 +1664,7 @@ func (s *Server) debugSubscribers(sub *subscription, c *client, subject, reply s
 		delete(s.sys.replies, replySubj)
 		s.mu.Unlock()
 		// Send the response.
-		s.sendInternalAccountMsg(c.acc, reply, atomic.LoadInt32(&nsubs))
+		s.sendInternalAccountMsg(nil, reply, atomic.LoadInt32(&nsubs))
 	}()
 }
 
diff --git a/server/jetstream_api.go b/server/jetstream_api.go
index ba12c443..8b34cb30 100644
--- a/server/jetstream_api.go
+++ b/server/jetstream_api.go
@@ -552,7 +552,7 @@ func (s *Server) setJetStreamExportSubs() error {
 }
 
 func (s *Server) sendAPIResponse(c *client, subject, reply, request, response string) {
-	s.sendInternalAccountMsg(c.acc, reply, response)
+	s.sendInternalAccountMsg(nil, reply, response)
 	s.sendJetStreamAPIAuditAdvisory(c, subject, request, response)
 }
 
diff --git a/server/server.go b/server/server.go
index 1811c8d3..37cbdf2c 100644
--- a/server/server.go
+++ b/server/server.go
@@ -422,9 +422,6 @@ func NewServer(opts *Options) (*Server, error) {
 				return nil
 			}
 			if _, ok := s.accounts.Load(accName); !ok {
-				if accName == DEFAULT_SYSTEM_ACCOUNT && opts.SystemAccount == _EMPTY_ && !opts.NoSystemAccount {
-					return nil
-				}
 				return fmt.Errorf("cannot find account %q specified in leafnode authorization", accName)
 			}
 			return nil
@@ -445,9 +442,6 @@ func NewServer(opts *Options) (*Server, error) {
 				continue
 			}
 			if _, ok := s.accounts.Load(r.LocalAccount); !ok {
-				if r.LocalAccount == DEFAULT_SYSTEM_ACCOUNT && opts.SystemAccount == _EMPTY_ && !opts.NoSystemAccount {
-					continue
-				}
 				return nil, fmt.Errorf("no local account %q for remote leafnode", r.LocalAccount)
 			}
 		}
diff --git a/server/stream.go b/server/stream.go
index 770fb012..2698080a 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -873,6 +873,7 @@ func (mset *Stream) processInboundJetStreamMsg(_ *subscription, pc *client, subj
 	if c != nil && c.acc != nil {
 		accName = c.acc.Name
 	}
+
 	doAck := !mset.config.NoAck
 	pubAck := mset.pubAck
 	jsa := mset.jsa
@@ -1099,6 +1100,7 @@ func (mset *Stream) internalSendLoop() {
 			didDeliver := c.processInboundClientMsg(msg)
 			c.pa.szb = nil
 			c.flushClients(0)
+
 			// Check to see if this is a delivery for an observable and
 			// we failed to deliver the message. If so alert the observable.
 			if pm.o != nil && pm.seq > 0 && !didDeliver {


Test patch:
diff --git a/server/accounts_test.go b/server/accounts_test.go
index 35d74277..ee98ab95 100644
--- a/server/accounts_test.go
+++ b/server/accounts_test.go
@@ -661,7 +661,7 @@ func TestSimpleMapping(t *testing.T) {
 		}
 		matches := mraw[0]
 		if matches[SUB_INDEX] != "import.foo" {
-			t.Fatalf("Did not get correct subject: '%s'", matches[SUB_INDEX])
+			t.Fatalf("Did not get correct subject: wanted %q, got %q", "import.foo", matches[SUB_INDEX])
 		}
 		if matches[SID_INDEX] != sid {
 			t.Fatalf("Did not get correct sid: '%s'", matches[SID_INDEX])
diff --git a/test/jetstream_test.go b/test/jetstream_test.go
index cea5fc77..06ade7fc 100644
--- a/test/jetstream_test.go
+++ b/test/jetstream_test.go
@@ -1099,7 +1099,7 @@ func TestJetStreamAddStreamSameConfigOK(t *testing.T) {
 	}
 }
 
-func sendStreamMsg(t *testing.T, nc *nats.Conn, subject, msg string) uint64 {
+func sendStreamMsg(t *testing.T, nc *nats.Conn, subject, msg string) *server.PubAck {
 	t.Helper()
 	resp, _ := nc.Request(subject, []byte(msg), 500*time.Millisecond)
 	if resp == nil {
@@ -1112,7 +1112,7 @@ func sendStreamMsg(t *testing.T, nc *nats.Conn, subject, msg string) uint64 {
 	if err := json.Unmarshal(resp.Data[3:], &pubAck); err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
-	return pubAck.Seq
+	return &pubAck
 }
 
 func TestJetStreamBasicAckPublish(t *testing.T) {
@@ -9800,3 +9800,147 @@ func TestJetStreamDeliveryAfterServerRestart(t *testing.T) {
 		t.Fatalf("Did not get message: %v", err)
 	}
 }
+
+// This is for the basics of importing the ability to send to a stream and consume
+// from a consumer that is pull based on push based on a well known delivery subject.
+func TestJetStreamAccountImportBasics(t *testing.T) {
+	conf := createConfFile(t, []byte(`
+		listen: 127.0.0.1:-1
+		no_auth_user: rip
+		jetstream: {max_mem_store: 64GB, max_file_store: 10TB}
+		accounts: {
+			JS: {
+				jetstream: enabled
+				users: [ {user: dlc, password: foo} ]
+				exports [
+					# This is for sending into a stream from other accounts.
+					{ service: "ORDERS.*" }
+					# This is for accessing a pull based consumer.
+					{ service: "$JS.API.CONSUMER.MSG.NEXT.*.*" }
+					# This is streaming to a delivery subject for a push based consumer.
+					{ stream: "deliver.ORDERS" }
+					# This is to ack received messages. This is a service to ack acks..
+					{ service: "$JS.ACK.ORDERS.*.>" }
+				]
+			},
+			IU: {
+				users: [ {user: rip, password: bar} ]
+				imports [
+					{ service: { subject: "ORDERS.*", account: JS }, to: "my.orders.$1" }
+					{ service: { subject: "$JS.API.CONSUMER.MSG.NEXT.ORDERS.d", account: JS }, to: "nxt.msg" }
+					{ stream:  { subject: "deliver.ORDERS", account: JS }, to: "d" }
+					{ service: { subject: "$JS.ACK.ORDERS.*.>", account: JS } }
+				]
+			},
+		}
+	`))
+	defer os.Remove(conf)
+
+	s, _ := RunServerWithConfig(conf)
+	defer s.Shutdown()
+
+	if config := s.JetStreamConfig(); config != nil {
+		defer os.RemoveAll(config.StoreDir)
+	}
+
+	acc, err := s.LookupAccount("JS")
+	if err != nil {
+		t.Fatalf("Unexpected error looking up account: %v", err)
+	}
+
+	mset, err := acc.AddStream(&server.StreamConfig{Name: "ORDERS", Subjects: []string{"ORDERS.*"}})
+	if err != nil {
+		t.Fatalf("Unexpected error adding stream: %v", err)
+	}
+	defer mset.Delete()
+
+	// This should be the rip user, the one that imports some JS.
+	nc := clientConnectToServer(t, s)
+	defer nc.Close()
+
+	// Simple publish to a stream.
+	pubAck := sendStreamMsg(t, nc, "my.orders.foo", "ORDERS-1")
+	if pubAck.Stream != "ORDERS" || pubAck.Seq != 1 {
+		t.Fatalf("Bad pubAck received: %+v", pubAck)
+	}
+	if msgs := mset.State().Msgs; msgs != 1 {
+		t.Fatalf("Expected 1 message, got %d", msgs)
+	}
+
+	total := 2
+	for i := 2; i <= total; i++ {
+		sendStreamMsg(t, nc, "my.orders.bar", fmt.Sprintf("ORDERS-%d", i))
+	}
+	if msgs := mset.State().Msgs; msgs != uint64(total) {
+		t.Fatalf("Expected %d messages, got %d", total, msgs)
+	}
+
+	// Now test access to a pull based consumer, e.g. workqueue.
+	o, err := mset.AddConsumer(&server.ConsumerConfig{
+		Durable:   "d",
+		AckPolicy: server.AckExplicit,
+	})
+	if err != nil {
+		t.Fatalf("Expected no error, got %v", err)
+	}
+	defer o.Delete()
+
+	// We mapped the next message request, "$JS.API.CONSUMER.MSG.NEXT.ORDERS.d" -> "nxt.msg"
+	m, err := nc.Request("nxt.msg", nil, time.Second)
+	if err != nil {
+		t.Fatalf("Unexpected error: %v", err)
+	}
+	if string(m.Data) != "ORDERS-1" {
+		t.Fatalf("Expected to receive %q, got %q", "ORDERS-1", m.Data)
+	}
+
+	// Now test access to a push based consumer
+	o, err = mset.AddConsumer(&server.ConsumerConfig{
+		Durable:        "p",
+		DeliverSubject: "deliver.ORDERS",
+		AckPolicy:      server.AckExplicit,
+	})
+	if err != nil {
+		t.Fatalf("Expected no error, got %v", err)
+	}
+	defer o.Delete()
+
+	// We remapped from above, deliver.ORDERS -> d
+	sub, _ := nc.SubscribeSync("d")
+	defer sub.Unsubscribe()
+
+	checkFor(t, 250*time.Millisecond, 10*time.Millisecond, func() error {
+		if nmsgs, _, _ := sub.Pending(); err != nil || nmsgs != total {
+			return fmt.Errorf("Did not receive correct number of messages: %d vs %d", nmsgs, total)
+		}
+		return nil
+	})
+
+	m, _ = sub.NextMsg(time.Second)
+	// Make sure we remapped subject correctly across the account boundary.
+	if m.Subject != "ORDERS.foo" {
+		t.Fatalf("Expected subject of %q, got %q", "ORDERS.foo", m.Subject)
+	}
+	// Now make sure we can ack messages correctly.
+	m.Ack()
+	nc.Flush()
+
+	if info := o.Info(); info.AckFloor.Consumer != 1 {
+		t.Fatalf("Did not receive the ack properly")
+	}
+
+	// Grab second one now.
+	m, _ = sub.NextMsg(time.Second)
+	// Make sure we remapped subject correctly across the account boundary.
+	if m.Subject != "ORDERS.bar" {
+		t.Fatalf("Expected subject of %q, got %q", "ORDERS.bar", m.Subject)
+	}
+	// Now make sure we can ack messages and get back an ack as well.
+	resp, _ := nc.Request(m.Reply, nil, 100*time.Millisecond)
+	if resp == nil {
+		t.Fatalf("No response, possible timeout?")
+	}
+	if info := o.Info(); info.AckFloor.Consumer != 2 {
+		t.Fatalf("Did not receive the ack properly")
+	}
+}