fixup for never micro

Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>

http.go

@@ -17,7 +17,6 @@ import (
 	"github.com/unistack-org/micro/v3/logger"
 	"github.com/unistack-org/micro/v3/registry"
 	"github.com/unistack-org/micro/v3/server"
-	regutil "github.com/unistack-org/micro/v3/util/registry"
 )
 
 var (
@@ -137,14 +136,11 @@ func (h *httpServer) Register() error {
 	eps := h.hd.Endpoints()
 	h.Unlock()
 
-	service, err := regutil.NewService(h)
+	service, err := server.NewRegistryService(h)
 	if err != nil {
 		return err
 	}
-	if len(service.Metadata) == 0 {
-		service.Metadata = make(map[string]string, 5)
-	}
-	service.Metadata["protocol"] = "http"
+	service.Nodes[0].Metadata["protocol"] = "http"
 	service.Endpoints = eps
 
 	h.Lock()
@@ -183,6 +179,8 @@ func (h *httpServer) Register() error {
 	}
 	h.registered = true
 
+	subCtx := h.opts.Context
+
 	for sb := range h.subscribers {
 		handler := h.createSubHandler(sb, opts)
 		var subOpts []broker.SubscribeOption
@@ -190,11 +188,15 @@ func (h *httpServer) Register() error {
 			subOpts = append(subOpts, broker.Queue(queue))
 		}
 
+		if cx := sb.Options().Context; cx != nil {
+			subCtx = cx
+		}
+
 		if !sb.Options().AutoAck {
 			subOpts = append(subOpts, broker.DisableAutoAck())
 		}
 
-		sub, err := opts.Broker.Subscribe(sb.Topic(), handler, subOpts...)
+		sub, err := opts.Broker.Subscribe(subCtx, sb.Topic(), handler, subOpts...)
 		if err != nil {
 			return err
 		}
@@ -210,7 +212,7 @@ func (h *httpServer) Deregister() error {
 
 	logger.Infof("Deregistering node: %s", opts.Name+"-"+opts.Id)
 
-	service, err := regutil.NewService(h)
+	service, err := server.NewRegistryService(h)
 	if err != nil {
 		return err
 	}
@@ -226,10 +228,17 @@ func (h *httpServer) Deregister() error {
 	}
 	h.registered = false
 
+	subCtx := h.opts.Context
 	for sb, subs := range h.subscribers {
 		for _, sub := range subs {
+			if cx := sb.Options().Context; cx != nil {
+				subCtx = cx
+			}
 			logger.Infof("Unsubscribing from topic: %s", sub.Topic())
-			sub.Unsubscribe()
+			if err := sub.Unsubscribe(subCtx); err != nil {
+				logger.Errorf("failed to unsubscribe topic: %s, error: %v", sb.Topic(), err)
+				return err
+			}
 		}
 		h.subscribers[sb] = nil
 	}
@@ -261,7 +270,7 @@ func (h *httpServer) Start() error {
 		return errors.New("Server required http.Handler")
 	}
 
-	if err = opts.Broker.Connect(); err != nil {
+	if err = opts.Broker.Connect(h.opts.Context); err != nil {
 		return err
 	}
 
@@ -334,7 +343,7 @@ func (h *httpServer) Start() error {
 		// deregister
 		h.Deregister()
 
-		opts.Broker.Disconnect()
+		opts.Broker.Disconnect(h.opts.Context)
 	}()
 
 	return nil