Refactor: ws

Signed-off-by: nyagamunene <stevenyaga2014@gmail.com>

Author: nyagamunene

ws/adapter.go

@@ -35,7 +35,7 @@ type Service interface {
 	// Subscribe subscribes message from the broker using the thingKey for authorization,
 	// and the channelID for subscription. Subtopic is optional.
 	// If the subscription is successful, nil is returned otherwise error is returned.
-	Subscribe(ctx context.Context, thingKey, chanID, subtopic string, client *Client) error
+	Subscribe(ctx context.Context, thingKey, chanID, subtopic string, client *Client) errors.Error
 }
 
 var _ Service = (*adapterService)(nil)
@@ -53,7 +53,7 @@ func New(authClient magistrala.AuthzServiceClient, pubsub messaging.PubSub) Serv
 	}
 }
 
-func (svc *adapterService) Subscribe(ctx context.Context, thingKey, chanID, subtopic string, c *Client) error {
+func (svc *adapterService) Subscribe(ctx context.Context, thingKey, chanID, subtopic string, c *Client) errors.Error {
 	if chanID == "" || thingKey == "" {
 		return svcerr.ErrAuthentication
 	}
@@ -84,7 +84,7 @@ func (svc *adapterService) Subscribe(ctx context.Context, thingKey, chanID, subt
 
 // authorize checks if the thingKey is authorized to access the channel
 // and returns the thingID if it is.
-func (svc *adapterService) authorize(ctx context.Context, thingKey, chanID, action string) (string, error) {
+func (svc *adapterService) authorize(ctx context.Context, thingKey, chanID, action string) (string, errors.Error) {
 	ar := &magistrala.AuthorizeReq{
 		SubjectType: auth.ThingType,
 		Permission:  action,

ws/api/logging.go

@@ -8,6 +8,7 @@ import (
 	"log/slog"
 	"time"
 
+	"github.com/absmach/magistrala/pkg/errors"
 	"github.com/absmach/magistrala/ws"
 )
 
@@ -25,7 +26,7 @@ func LoggingMiddleware(svc ws.Service, logger *slog.Logger) ws.Service {
 
 // Subscribe logs the subscribe request. It logs the channel and subtopic(if present) and the time it took to complete the request.
 // If the request fails, it logs the error.
-func (lm *loggingMiddleware) Subscribe(ctx context.Context, thingKey, chanID, subtopic string, c *ws.Client) (err error) {
+func (lm *loggingMiddleware) Subscribe(ctx context.Context, thingKey, chanID, subtopic string, c *ws.Client) (err errors.Error) {
 	defer func(begin time.Time) {
 		args := []any{
 			slog.String("duration", time.Since(begin).String()),

ws/api/metrics.go

@@ -9,6 +9,7 @@ import (
 	"context"
 	"time"
 
+	"github.com/absmach/magistrala/pkg/errors"
 	"github.com/absmach/magistrala/ws"
 	"github.com/go-kit/kit/metrics"
 )
@@ -31,7 +32,7 @@ func MetricsMiddleware(svc ws.Service, counter metrics.Counter, latency metrics.
 }
 
 // Subscribe instruments Subscribe method with metrics.
-func (mm *metricsMiddleware) Subscribe(ctx context.Context, thingKey, chanID, subtopic string, c *ws.Client) error {
+func (mm *metricsMiddleware) Subscribe(ctx context.Context, thingKey, chanID, subtopic string, c *ws.Client) errors.Error {
 	defer func(begin time.Time) {
 		mm.counter.With("method", "subscribe").Add(1)
 		mm.latency.With("method", "subscribe").Observe(time.Since(begin).Seconds())

ws/client.go

@@ -4,6 +4,7 @@
 package ws
 
 import (
+	"github.com/absmach/magistrala/pkg/errors"
 	"github.com/absmach/magistrala/pkg/messaging"
 	"github.com/gorilla/websocket"
 )
@@ -23,19 +24,19 @@ func NewClient(c *websocket.Conn) *Client {
 }
 
 // Cancel handles the websocket connection after unsubscribing.
-func (c *Client) Cancel() error {
+func (c *Client) Cancel() errors.Error {
 	if c.conn == nil {
 		return nil
 	}
-	return c.conn.Close()
+	return errors.Cast(c.conn.Close())
 }
 
 // Handle handles the sending and receiving of messages via the broker.
-func (c *Client) Handle(msg *messaging.Message) error {
+func (c *Client) Handle(msg *messaging.Message) errors.Error {
 	// To prevent publisher from receiving its own published message
 	if msg.GetPublisher() == c.id {
 		return nil
 	}
 
-	return c.conn.WriteMessage(websocket.TextMessage, msg.GetPayload())
+	return errors.Cast(c.conn.WriteMessage(websocket.TextMessage, msg.GetPayload()))
 }

ws/tracing/tracing.go

@@ -6,6 +6,7 @@ package tracing
 import (
 	"context"
 
+	"github.com/absmach/magistrala/pkg/errors"
 	"github.com/absmach/magistrala/ws"
 	"go.opentelemetry.io/otel/trace"
 )
@@ -32,7 +33,7 @@ func New(tracer trace.Tracer, svc ws.Service) ws.Service {
 }
 
 // Subscribe traces the "Subscribe" operation of the wrapped ws.Service.
-func (tm *tracingMiddleware) Subscribe(ctx context.Context, thingKey, chanID, subtopic string, client *ws.Client) error {
+func (tm *tracingMiddleware) Subscribe(ctx context.Context, thingKey, chanID, subtopic string, client *ws.Client) errors.Error {
 	ctx, span := tm.tracer.Start(ctx, subscribeOP)
 	defer span.End()