coap-gw: check for offline event when coap-gw has been shutdown

coap-gateway/service/service_test.go

@@ -0,0 +1,96 @@
+package service_test
+
+import (
+	"context"
+	"crypto/tls"
+	"testing"
+
+	coapgwTest "github.com/plgd-dev/hub/v2/coap-gateway/test"
+	"github.com/plgd-dev/hub/v2/grpc-gateway/client"
+	"github.com/plgd-dev/hub/v2/grpc-gateway/pb"
+	kitNetGrpc "github.com/plgd-dev/hub/v2/pkg/net/grpc"
+	test "github.com/plgd-dev/hub/v2/test"
+	"github.com/plgd-dev/hub/v2/test/config"
+	oauthTest "github.com/plgd-dev/hub/v2/test/oauth-server/test"
+	"github.com/plgd-dev/hub/v2/test/service"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/atomic"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
+)
+
+type testDevsObs struct {
+	err atomic.Error
+	ch  chan client.DevicesObservationEvent
+}
+
+func (t *testDevsObs) Error(err error) {
+	t.err.Store(err)
+}
+
+func (t *testDevsObs) Handle(ctx context.Context, event client.DevicesObservationEvent) error {
+	t.ch <- event
+	return nil
+}
+
+func (t *testDevsObs) OnClose() {}
+
+func TestShutdownServiceWithDeviceIssue627(t *testing.T) {
+	deviceID := test.MustFindDeviceByName(test.TestDeviceName)
+	ctx, cancel := context.WithTimeout(context.Background(), config.TEST_TIMEOUT)
+	defer cancel()
+
+	const services = service.SetUpServicesOAuth | service.SetUpServicesId | service.SetUpServicesResourceDirectory |
+		service.SetUpServicesGrpcGateway | service.SetUpServicesResourceAggregate
+	tearDown := service.SetUpServices(ctx, t, services)
+	defer tearDown()
+	// log.Setup(log.Config{Debug: true})
+
+	ctx = kitNetGrpc.CtxWithToken(ctx, oauthTest.GetDefaultAccessToken(t))
+
+	coapShutdown := coapgwTest.SetUp(t)
+	defer coapShutdown()
+
+	grpcConn, err := grpc.Dial(config.GRPC_HOST, grpc.WithTransportCredentials(credentials.NewTLS(&tls.Config{
+		RootCAs: test.GetRootCertificatePool(t),
+	})))
+	require.NoError(t, err)
+	defer func() {
+		_ = grpcConn.Close()
+	}()
+	grpcClient := client.New(pb.NewGrpcGatewayClient(grpcConn))
+
+	_, shutdownDevSim := test.OnboardDevSim(ctx, t, pb.NewGrpcGatewayClient(grpcConn), deviceID, config.GW_HOST, test.GetAllBackendResourceLinks())
+	defer shutdownDevSim()
+
+	ch := make(chan client.DevicesObservationEvent, 1000)
+
+	v := testDevsObs{
+		ch: ch,
+	}
+
+	observationID, err := grpcClient.ObserveDevices(ctx, &v)
+	require.NoError(t, err)
+	defer func(observationID string) {
+		err := grpcClient.StopObservingDevices(ctx, observationID)
+		require.NoError(t, err)
+		require.NoError(t, v.err.Load())
+	}(observationID)
+
+	coapShutdown()
+
+	for {
+		select {
+		case e := <-ch:
+			if e.Event != client.DevicesObservationEvent_OFFLINE {
+				continue
+			}
+			require.Len(t, e.DeviceIDs, 1)
+			require.Equal(t, deviceID, e.DeviceIDs[0])
+			return
+		case <-ctx.Done():
+			require.NoError(t, ctx.Err())
+		}
+	}
+
+}

grpc-gateway/client/observeDevices.go

@@ -2,7 +2,6 @@ package client
 
 import (
 	"context"
-	"fmt"
 
 	"github.com/google/uuid"
 	"github.com/plgd-dev/hub/v2/grpc-gateway/pb"
@@ -77,9 +76,9 @@ func (c *Client) ObserveDevices(ctx context.Context, handler DevicesObservationH
 	sub, err := c.NewDevicesSubscription(ctx, &devicesObservation{
 		h: handler,
 		removeSubscription: func() {
-			if _, err := c.stopObservingDevices(ID.String()); err != nil {
-				handler.Error(fmt.Errorf("failed to stop device('%v') observation: %w", ID.String(), err))
-			}
+			// we can ignore err during removeSubscription, if err != nil then some other
+			// part of code already removed the subscription
+			_, _ = c.stopObservingDevices(ID.String())
 		},
 	})
 	if err != nil {