MF-898 - Change thing's service to use bulk connect

bootstrap/mocks/things.go

@@ -70,7 +70,7 @@ func (svc *mainfluxThings) ViewThing(_ context.Context, owner, id string) (thing
 	return things.Thing{}, things.ErrNotFound
 }
 
-func (svc *mainfluxThings) Connect(_ context.Context, owner, chanID, thingID string) error {
+func (svc *mainfluxThings) Connect(_ context.Context, owner, chID string, thIDs ...string) error {
 	svc.mu.Lock()
 	defer svc.mu.Unlock()
 
@@ -79,11 +79,13 @@ func (svc *mainfluxThings) Connect(_ context.Context, owner, chanID, thingID str
 		return things.ErrUnauthorizedAccess
 	}
 
-	if svc.channels[chanID].Owner != userID.Value {
-		return things.ErrNotFound
-	}
+	for _, thID := range thIDs {
+		if svc.channels[chID].Owner != userID.Value {
+			return things.ErrNotFound
+		}
 
-	svc.connections[chanID] = append(svc.connections[chanID], thingID)
+		svc.connections[chID] = append(svc.connections[chID], thID)
+	}
 	return nil
 }
 

things/api/logging.go

@@ -201,17 +201,17 @@ func (lm *loggingMiddleware) RemoveChannel(ctx context.Context, token, id string
 	return lm.svc.RemoveChannel(ctx, token, id)
 }
 
-func (lm *loggingMiddleware) Connect(ctx context.Context, token, chanID, thingID string) (err error) {
+func (lm *loggingMiddleware) Connect(ctx context.Context, token, chID string, thIDs ...string) (err error) {
 	defer func(begin time.Time) {
-		message := fmt.Sprintf("Method connect for token %s, channel %s and thing %s took %s to complete", token, chanID, thingID, time.Since(begin))
+		message := fmt.Sprintf("Method connect for token %s, channel %s and things %s took %s to complete", token, chID, thIDs, time.Since(begin))
 		if err != nil {
 			lm.logger.Warn(fmt.Sprintf("%s with error: %s.", message, err))
 			return
 		}
 		lm.logger.Info(fmt.Sprintf("%s without errors.", message))
 	}(time.Now())
 
-	return lm.svc.Connect(ctx, token, chanID, thingID)
+	return lm.svc.Connect(ctx, token, chID, thIDs...)
 }
 
 func (lm *loggingMiddleware) Disconnect(ctx context.Context, token, chanID, thingID string) (err error) {

things/api/metrics.go

@@ -148,13 +148,13 @@ func (ms *metricsMiddleware) RemoveChannel(ctx context.Context, token, id string
 	return ms.svc.RemoveChannel(ctx, token, id)
 }
 
-func (ms *metricsMiddleware) Connect(ctx context.Context, token, chanID, thingID string) error {
+func (ms *metricsMiddleware) Connect(ctx context.Context, token, chanID string, thIDs ...string) error {
 	defer func(begin time.Time) {
 		ms.counter.With("method", "connect").Add(1)
 		ms.latency.With("method", "connect").Observe(time.Since(begin).Seconds())
 	}(time.Now())
 
-	return ms.svc.Connect(ctx, token, chanID, thingID)
+	return ms.svc.Connect(ctx, token, chanID, thIDs...)
 }
 
 func (ms *metricsMiddleware) Disconnect(ctx context.Context, token, chanID, thingID string) error {

things/redis/streams.go

@@ -186,21 +186,23 @@ func (es eventStore) RemoveChannel(ctx context.Context, token, id string) error
 	return nil
 }
 
-func (es eventStore) Connect(ctx context.Context, token, chanID, thingID string) error {
-	if err := es.svc.Connect(ctx, token, chanID, thingID); err != nil {
+func (es eventStore) Connect(ctx context.Context, token, chID string, thIDs ...string) error {
+	if err := es.svc.Connect(ctx, token, chID, thIDs...); err != nil {
 		return err
 	}
 
-	event := connectThingEvent{
-		chanID:  chanID,
-		thingID: thingID,
-	}
-	record := &redis.XAddArgs{
-		Stream:       streamID,
-		MaxLenApprox: streamLen,
-		Values:       event.Encode(),
+	for _, thID := range thIDs {
+		event := connectThingEvent{
+			chanID:  chID,
+			thingID: thID,
+		}
+		record := &redis.XAddArgs{
+			Stream:       streamID,
+			MaxLenApprox: streamLen,
+			Values:       event.Encode(),
+		}
+		es.client.XAdd(record).Err()
 	}
-	es.client.XAdd(record).Err()
 
 	return nil
 }

things/service.go

@@ -84,8 +84,8 @@ type Service interface {
 	// belongs to the user identified by the provided key.
 	RemoveChannel(context.Context, string, string) error
 
-	// Connect adds thing to the channel's list of connected things.
-	Connect(context.Context, string, string, string) error
+	// Connect adds things to the channel's list of connected things.
+	Connect(context.Context, string, string, ...string) error
 
 	// Disconnect removes thing from the channel's list of connected
 	// things.
@@ -283,13 +283,13 @@ func (ts *thingsService) RemoveChannel(ctx context.Context, token, id string) er
 	return ts.channels.Remove(ctx, res.GetValue(), id)
 }
 
-func (ts *thingsService) Connect(ctx context.Context, token, chanID, thingID string) error {
+func (ts *thingsService) Connect(ctx context.Context, token, chID string, thIDs ...string) error {
 	res, err := ts.users.Identify(ctx, &mainflux.Token{Value: token})
 	if err != nil {
 		return ErrUnauthorizedAccess
 	}
 
-	return ts.channels.Connect(ctx, res.GetValue(), chanID, thingID)
+	return ts.channels.Connect(ctx, res.GetValue(), chID, thIDs...)
 }
 
 func (ts *thingsService) Disconnect(ctx context.Context, token, chanID, thingID string) error {