trying to track down this data race and determine if its my code or n…

…ats.go

README.md

@@ -93,7 +93,7 @@ func main() {
 	if err != nil {
 		log.Fatal(err)
 	}
-	go ns.Start()
+	ns.Start()
 	defer func() {
 		ns.Shutdown()
 		ns.WaitForShutdown()

examples/simple/server/main.go

@@ -33,7 +33,7 @@ func main() {
 	if err != nil {
 		log.Fatal(err)
 	}
-	go ns.Start()
+	ns.Start()
 	defer func() {
 		ns.Shutdown()
 		ns.WaitForShutdown()

server.go

@@ -40,14 +40,16 @@ func (s *ServerConfig) setDefaults() {
 
 // Server represents a stormRPC server. It contains all functionality for handling RPC requests.
 type Server struct {
-	mu             sync.Mutex
-	nc             *nats.Conn
+	mu sync.Mutex
+	// nc             *nats.Conn
 	shutdownSignal chan struct{}
 	handlerFuncs   map[string]HandlerFunc
 	errorHandler   ErrorHandler
 	timeout        time.Duration
 	mw             []Middleware
 
+	running bool
+
 	svc micro.Service
 }
 
@@ -81,12 +83,23 @@ func NewServer(cfg *ServerConfig, opts ...ServerOption) (*Server, error) {
 		return nil, err
 	}
 
+	err = svc.AddEndpoint("123123123123123", micro.HandlerFunc(func(r micro.Request) {}))
+	if err != nil {
+		return nil, err
+	}
+
+	err = svc.AddEndpoint("0909090", micro.HandlerFunc(func(r micro.Request) {}))
+	if err != nil {
+		return nil, err
+	}
+
 	return &Server{
-		nc:             nc,
+		// nc:             nc,
 		shutdownSignal: make(chan struct{}),
 		handlerFuncs:   make(map[string]HandlerFunc),
 		timeout:        defaultServerTimeout,
 		errorHandler:   cfg.errorHandler,
+		running:        false,
 		svc:            svc,
 	}, nil
 }
@@ -128,12 +141,21 @@ func (s *Server) Handle(subject string, fn HandlerFunc) {
 func (s *Server) Run() error {
 	s.mu.Lock()
 	s.applyMiddlewares()
-	for sub, fn := range s.handlerFuncs {
-		if err := s.createMicroEndpoint(sub, fn); err != nil {
+	s.mu.Unlock()
+	for sub := range s.handlerFuncs {
+		if err := s.createMicroEndpoint(s.svc, sub, s.handlerFuncs[sub]); err != nil {
 			return err
 		}
 	}
+	// for sub, fn := range s.handlerFuncs {
+	// 	if err := s.createMicroEndpoint(sub, fn); err != nil {
+	// 		return err
+	// 	}
+	// }
+	s.mu.Lock()
+	s.running = true
 	s.mu.Unlock()
+	// time.Sleep(1 * time.Second)
 
 	<-s.shutdownSignal
 	return nil
@@ -143,15 +165,16 @@ func (s *Server) Run() error {
 func (s *Server) Shutdown(ctx context.Context) error {
 	s.mu.Lock()
 	defer s.mu.Unlock()
-	if err := s.nc.FlushWithContext(ctx); err != nil {
-		return err
-	}
+	// if err := s.nc.FlushWithContext(ctx); err != nil {
+	// 	return err
+	// }
 
 	if err := s.svc.Stop(); err != nil {
 		return err
 	}
 
-	s.nc.Close()
+	// s.nc.Close()
+	s.running = false
 	s.shutdownSignal <- struct{}{}
 	return nil
 }
@@ -174,7 +197,9 @@ func (s *Server) Use(mw ...Middleware) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 
-	s.mw = mw
+	if !s.running {
+		s.mw = append(s.mw, mw...)
+	}
 }
 
 func (s *Server) applyMiddlewares() {
@@ -189,7 +214,9 @@ func (s *Server) applyMiddlewares() {
 
 // createMicroEndpoint registers a HandlerFunc as a micro Endpoint
 // allowing for automatic service discovery and observability.
-func (s *Server) createMicroEndpoint(subject string, handlerFunc HandlerFunc) error {
+func (s *Server) createMicroEndpoint(svc micro.Service, subject string, handlerFunc HandlerFunc) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
 	err := s.svc.AddEndpoint(
 		nameFromSubject(subject),
 		micro.ContextHandler(context.Background(), func(ctx context.Context, r micro.Request) {
@@ -210,10 +237,8 @@ func (s *Server) createMicroEndpoint(subject string, handlerFunc HandlerFunc) er
 			resp := handlerFunc(ctx, Request{
 				Msg: &nats.Msg{
 					Subject: r.Subject(),
-					Reply:   "",
 					Header:  nats.Header(r.Headers()),
 					Data:    r.Data(),
-					Sub:     &nats.Subscription{},
 				},
 			})