@@ -5,71 +5,81 @@ import (
55 "fmt"
66
77 "github.com/facebookincubator/go-belt/tool/logger"
8- "github.com/scorfly/gokick"
8+ "github.com/xaionaro-go/chatwebhook/pkg/chatwebhook/kickcom"
9+ chatwebhookclient "github.com/xaionaro-go/chatwebhook/pkg/grpc/client"
10+ "github.com/xaionaro-go/chatwebhook/pkg/grpc/protobuf/go/chatwebhook_grpc"
11+ "github.com/xaionaro-go/observability"
12+ "github.com/xaionaro-go/streamctl/pkg/streamcontrol"
13+ "github.com/xaionaro-go/streamctl/pkg/streamcontrol/protobuf/goconv"
914)
1015
16+ type ChatHandlerAbstract interface {
17+ GetMessagesChan (
18+ ctx context.Context ,
19+ ) (<- chan streamcontrol.Event , error )
20+ }
21+
1122type ChatHandler struct {
12- onClose func (context. Context )
23+ Client * chatwebhookclient. Client
1324}
1425
26+ var _ ChatHandlerAbstract = (* ChatHandler )(nil )
27+
1528func NewChatHandler (
1629 ctx context.Context ,
17- client * gokick.Client ,
18- broadcasterUserID * int ,
19- onClose func (context.Context ),
30+ client * chatwebhookclient.Client ,
2031) (_ret * ChatHandler , _err error ) {
21- logger .Debugf (ctx , "NewChatHandler(ctx, client, %p)" , onClose )
32+ logger .Debugf (ctx , "NewChatHandler" )
2233 defer func () {
23- logger .Debugf (ctx , "/NewChatHandler(ctx, client, %p) : %#+v %v" , onClose , _ret , _err )
34+ logger .Debugf (ctx , "/NewChatHandler: %#+v %v" , _ret , _err )
2435 }()
2536
37+ return & ChatHandler {
38+ Client : client ,
39+ }, nil
40+ }
41+
42+ func (h * ChatHandler ) GetMessagesChan (
43+ ctx context.Context ,
44+ ) (<- chan streamcontrol.Event , error ) {
2645 ctx , cancelFn := context .WithCancel (ctx )
2746 defer cancelFn ()
2847
29- subscriptions := []gokick.SubscriptionRequest {
30- {
31- Name : gokick .SubscriptionNameChatMessage ,
32- Version : 1 ,
33- },
34- {
35- Name : gokick .SubscriptionNameChannelFollow ,
36- Version : 1 ,
37- },
38- {
39- Name : gokick .SubscriptionNameChannelSubscriptionRenewal ,
40- Version : 1 ,
41- },
42- {
43- Name : gokick .SubscriptionNameChannelSubscriptionGifts ,
44- Version : 1 ,
45- },
46- {
47- Name : gokick .SubscriptionNameChannelSubscriptionCreated ,
48- Version : 1 ,
49- },
50- {
51- Name : gokick .SubscriptionNameLivestreamStatusUpdated ,
52- Version : 1 ,
53- },
54- {
55- Name : gokick .SubscriptionNameLivestreamMetadataUpdated ,
56- Version : 1 ,
57- },
58- {
59- Name : gokick .SubscriptionNameModerationBanned ,
60- Version : 1 ,
61- },
62- }
63- response , err := client .CreateSubscriptions (
64- ctx ,
65- gokick .SubscriptionMethodWebhook , // PANIC: we need websockets, we cannot use webhook
66- subscriptions ,
67- broadcasterUserID ,
68- )
48+ inCh , err := h .Client .GetMessagesChan (ctx , kickcom .ID , "" )
6949 if err != nil {
70- return nil , fmt .Errorf ("unable to create the event subscriptions: %w" , err )
50+ return nil , fmt .Errorf ("kick: failed to get messages chan: %w" , err )
51+ }
52+
53+ outCh := make (chan streamcontrol.Event , 1 )
54+ observability .Go (ctx , func (ctx context.Context ) {
55+ defer close (outCh )
56+ for {
57+ select {
58+ case <- ctx .Done ():
59+ return
60+ case ev , ok := <- inCh :
61+ if ! ok {
62+ return
63+ }
64+ msg , err := convertKickEventToChatMessage (ev )
65+ if err != nil {
66+ logger .Errorf (ctx , "failed to convert kick event to chat message: %v" , err )
67+ continue
68+ }
69+ outCh <- msg
70+ }
71+ }
72+ })
73+
74+ return outCh , nil
75+ }
76+
77+ func convertKickEventToChatMessage (
78+ ev * chatwebhook_grpc.Event ,
79+ ) (streamcontrol.Event , error ) {
80+ if ev == nil {
81+ return streamcontrol.Event {}, fmt .Errorf ("event is nil" )
7182 }
72- logger .Debugf (ctx , "subscriptions: %#+v" , response )
7383
74- panic ( "not implemented, yet" )
84+ return goconv . EventGRPC2Go ( ev ), nil
7585}
0 commit comments