Repository: nats-io/nats.go
Pull request #1872 (2025-05-13 22:48:48)

[FIXED] Stopping micro service should fully unsubscribe (#1871)
Fixes the issue describe in #1871  by ensuring the ranging of endpoints is not impacted by the action to perform the stop on the endpoint (which manipulates the endpoints slice). This is achieved by making a copy of the slice beforehand.

coveralls: 
[![Coverage Status](https://coveralls.io/builds/73655324/badge)](https://coveralls.io/builds/73655324)

coverage: 84.717% (+0.01%) from 84.703%
when pulling **62defadf1d32360ab3ffd4c582ec05e8b18a6bc3 on arunsworld:main**
into **70300b2e9480965e567f465acc26acb18efbc046 on nats-io:main**.


Patch:
diff --git a/kv.go b/kv.go
index dc401d5..e1ab3c0 100644
--- a/kv.go
+++ b/kv.go
@@ -1051,7 +1051,10 @@ func (kv *kvs) WatchFiltered(keys []string, opts ...WatchOpt) (KeyWatcher, error
 				w.initPending = delta
 			}
 			if w.received > w.initPending || delta == 0 {
-				w.initDoneTimer.Stop()
+				// Avoid possible race setting up timer.
+				if w.initDoneTimer != nil {
+					w.initDoneTimer.Stop()
+				}
 				w.initDone = true
 				w.updates <- nil
 			} else if w.initDoneTimer != nil {
diff --git a/micro/service.go b/micro/service.go
index ab75779..354f3d3 100644
--- a/micro/service.go
+++ b/micro/service.go
@@ -692,7 +692,10 @@ func (s *service) Stop() error {
 	if s.stopped {
 		return nil
 	}
-	for _, e := range s.endpoints {
+	// make a copy of s.endpoints to range over in order to stop
+	// since *Endpoint.stop manipulates s.endpoints!
+	endpointsToStop := append(make([]*Endpoint, 0, len(s.endpoints)), s.endpoints...)
+	for _, e := range endpointsToStop {
 		if err := e.stop(); err != nil {
 			fmt.Println("Error stopping endpoint: ", err)
 			return err
diff --git a/nats.go b/nats.go
index 1496fea..c032402 100644
--- a/nats.go
+++ b/nats.go
@@ -2292,6 +2292,21 @@ func (nc *Conn) ConnectedAddr() string {
 	return nc.conn.RemoteAddr().String()
 }
 
+// LocalAddr returns the local network address of the connection
+func (nc *Conn) LocalAddr() string {
+	if nc == nil {
+		return _EMPTY_
+	}
+
+	nc.mu.RLock()
+	defer nc.mu.RUnlock()
+
+	if nc.status != CONNECTED {
+		return _EMPTY_
+	}
+	return nc.conn.LocalAddr().String()
+}
+
 // ConnectedServerId reports the connected server's Id
 func (nc *Conn) ConnectedServerId() string {
 	if nc == nil {


Test patch:
diff --git a/test/nats_test.go b/test/nats_test.go
index 91ed88d..15bf492 100644
--- a/test/nats_test.go
+++ b/test/nats_test.go
@@ -553,6 +553,39 @@ func TestConnectedAddr(t *testing.T) {
 	}
 }
 
+func TestLocalAddr(t *testing.T) {
+	s := RunServerOnPort(TEST_PORT)
+	defer s.Shutdown()
+
+	var nc *nats.Conn
+	if addr := nc.LocalAddr(); addr != "" {
+		t.Fatalf("Expected empty result for nil connection, got %q", addr)
+	}
+	nc, err := nats.Connect(fmt.Sprintf("localhost:%d", TEST_PORT))
+	if err != nil {
+		t.Fatalf("Error connecting: %v", err)
+	}
+	addr := nc.LocalAddr()
+	if addr == "" {
+		t.Fatalf("Expected non-empty local address")
+	}
+	// Verify it's a valid address format
+	host, port, err := net.SplitHostPort(addr)
+	if err != nil {
+		t.Fatalf("Expected valid host:port format, got %q: %v", addr, err)
+	}
+	if host == "" {
+		t.Fatalf("Expected non-empty host in address %q", addr)
+	}
+	if port == "" {
+		t.Fatalf("Expected non-empty port in address %q", addr)
+	}
+	nc.Close()
+	if addr := nc.LocalAddr(); addr != "" {
+		t.Fatalf("Expected empty result for closed connection, got %q", addr)
+	}
+}
+
 func TestSubscribeSyncRace(t *testing.T) {
 	s := RunServerOnPort(TEST_PORT)
 	defer s.Shutdown()