Add a keepalive loop to the Kick handler

Author: xaionaro

pkg/streamcontrol/kick/client.go

@@ -12,6 +12,7 @@ type Client interface {
 	OnUserAccessTokenRefreshed(callback func(accessToken, refreshToken string))
 	UpdateStreamTitle(ctx context.Context, title string) (gokick.EmptyResponse, error)
 	UpdateStreamCategory(ctx context.Context, categoryID int) (gokick.EmptyResponse, error)
+	GetLivestreams(ctx context.Context, filter gokick.LivestreamListFilter) (gokick.LivestreamsResponseWrapper, error)
 	GetChannels(ctx context.Context, filter gokick.ChannelListFilter) (gokick.ChannelsResponseWrapper, error)
 	SendChatMessage(
 		ctx context.Context,

pkg/streamcontrol/kick/client_mock.go

@@ -32,6 +32,10 @@ func (clientMock) UpdateStreamCategory(ctx context.Context, categoryID int) (gok
 	return gokick.EmptyResponse{}, nil
 }
 
+func (clientMock) GetLivestreams(ctx context.Context, filter gokick.LivestreamListFilter) (gokick.LivestreamsResponseWrapper, error) {
+	return gokick.LivestreamsResponseWrapper{}, nil
+}
+
 func (clientMock) GetChannels(ctx context.Context, filter gokick.ChannelListFilter) (gokick.ChannelsResponseWrapper, error) {
 	return gokick.ChannelsResponseWrapper{
 		Result: []gokick.ChannelResponse{{

pkg/streamcontrol/kick/kick.go

@@ -82,6 +82,9 @@ func New(
 	}
 	k.SetClient(client)
 	client.OnUserAccessTokenRefreshed(k.onUserAccessTokenRefreshed)
+	observability.Go(ctx, func(ctx context.Context) {
+		k.keepAliveLoop(ctx)
+	})
 	return k, nil
 }
 
@@ -122,6 +125,28 @@ func (k *Kick) onUserAccessTokenRefreshed(
 	})
 }
 
+func (k *Kick) keepAliveLoop(
+	ctx context.Context,
+) {
+	logger.Debugf(ctx, "keepAliveLoop")
+	defer func() { logger.Debugf(ctx, "/keepAliveLoop") }()
+
+	t := time.NewTicker(time.Minute)
+	defer t.Stop()
+	for {
+		_, err := k.GetClient().GetLivestreams(k.CloseCtx, gokick.NewLivestreamListFilter().SetBroadcasterUserIDs(int(k.Channel.UserID)))
+		if err != nil {
+			logger.Errorf(ctx, "unable to get my stream status: %v", err)
+			continue
+		}
+		select {
+		case <-k.CloseCtx.Done():
+			return
+		case <-t.C:
+		}
+	}
+}
+
 func (k *Kick) initChatHandler(
 	ctx context.Context,
 ) error {
@@ -143,9 +168,12 @@ func (k *Kick) initChatHandler(
 			logger.Errorf(ctx, "unable to initialize chat handler: %v", err)
 			time.Sleep(time.Second)
 			select {
-			case <-ctx.Done():
+			case <-k.CloseCtx.Done():
 				logger.Debugf(ctx, "initChatHandler: cancelled (case #1)")
 				return
+			case <-ctx.Done():
+				logger.Debugf(ctx, "initChatHandler: cancelled (case #2)")
+				return
 			default:
 			}
 			chatHandler, err = k.newChatHandlerOBSOLETE(ctx, k.CurrentConfig.Config.Channel, k.onChatHandlerClose)
@@ -167,12 +195,17 @@ func (k *Kick) onChatHandlerClose(
 	defer k.ChatHandlerLocker.Unlock()
 	select {
 	case <-ctx.Done():
+		return
+	case <-k.CloseCtx.Done():
+		return
 	case <-time.After(time.Second):
 	}
 	for {
 		select {
 		case <-ctx.Done():
 			return
+		case <-k.CloseCtx.Done():
+			return
 		default:
 		}
 		chatHandler, err := k.newChatHandlerOBSOLETE(ctx, k.CurrentConfig.Config.Channel, k.onChatHandlerClose)
@@ -517,6 +550,8 @@ func (k *Kick) GetChatMessagesChan(
 			}()
 			logger.Tracef(ctx, "GetChatMessagesChan: waiting for a message")
 			select {
+			case <-k.CloseCtx.Done():
+				return
 			case <-ctx.Done():
 				return
 			case ev, ok := <-chatHandler.MessagesChan():