internal/processing/streaming/streamstatus.go


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50

package streaming

import (
	"encoding/json"
	"errors"
	"fmt"

	"github.com/sirupsen/logrus"
	apimodel "github.com/superseriousbusiness/gotosocial/internal/api/model"
	"github.com/superseriousbusiness/gotosocial/internal/gtsmodel"
)

func (p *processor) StreamStatusToAccount(s *apimodel.Status, account *gtsmodel.Account) error {
	l := p.log.WithFields(logrus.Fields{
		"func":    "StreamStatusForAccount",
		"account": account.ID,
	})
	v, ok := p.streamMap.Load(account.ID)
	if !ok {
		// no open connections so nothing to stream
		return nil
	}

	streamsForAccount, ok := v.(*gtsmodel.StreamsForAccount)
	if !ok {
		return errors.New("stream map error")
	}

	statusBytes, err := json.Marshal(s)
	if err != nil {
		return fmt.Errorf("error marshalling status to json: %s", err)
	}

	streamsForAccount.Lock()
	defer streamsForAccount.Unlock()
	for _, stream := range streamsForAccount.Streams {
		stream.Lock()
		defer stream.Unlock()
		if stream.Connected {
			l.Debugf("streaming status to stream id %s", stream.ID)
			stream.Messages <- &gtsmodel.Message{
				Stream:  []string{stream.Type},
				Event:   "update",
				Payload: string(statusBytes),
			}
		}
	}

	return nil
}