diff options
Diffstat (limited to 'internal/processing/streaming')
| -rw-r--r-- | internal/processing/streaming/authorize.go | 20 | ||||
| -rw-r--r-- | internal/processing/streaming/authorize_test.go | 48 | ||||
| -rw-r--r-- | internal/processing/streaming/notification.go | 37 | ||||
| -rw-r--r-- | internal/processing/streaming/notification_test.go | 60 | ||||
| -rw-r--r-- | internal/processing/streaming/openstream.go | 18 | ||||
| -rw-r--r-- | internal/processing/streaming/openstream_test.go | 41 | ||||
| -rw-r--r-- | internal/processing/streaming/streamdelete.go | 59 | ||||
| -rw-r--r-- | internal/processing/streaming/streaming.go | 33 | ||||
| -rw-r--r-- | internal/processing/streaming/streaming_test.go | 55 | ||||
| -rw-r--r-- | internal/processing/streaming/streamnotification.go | 51 | ||||
| -rw-r--r-- | internal/processing/streaming/streamstatus.go | 51 | ||||
| -rw-r--r-- | internal/processing/streaming/streamtoaccount.go | 55 | ||||
| -rw-r--r-- | internal/processing/streaming/update.go | 37 |
13 files changed, 420 insertions, 145 deletions
diff --git a/internal/processing/streaming/authorize.go b/internal/processing/streaming/authorize.go index f938a0c0c..1a5724f51 100644 --- a/internal/processing/streaming/authorize.go +++ b/internal/processing/streaming/authorize.go @@ -1,3 +1,21 @@ +/* + GoToSocial + Copyright (C) 2021 GoToSocial Authors admin@gotosocial.org + + This program is free software: you can redistribute it and/or modify + it under the terms of the GNU Affero General Public License as published by + the Free Software Foundation, either version 3 of the License, or + (at your option) any later version. + + This program is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU Affero General Public License for more details. + + You should have received a copy of the GNU Affero General Public License + along with this program. If not, see <http://www.gnu.org/licenses/>. +*/ + package streaming import ( @@ -8,7 +26,7 @@ import ( ) func (p *processor) AuthorizeStreamingRequest(ctx context.Context, accessToken string) (*gtsmodel.Account, error) { - ti, err := p.oauthServer.LoadAccessToken(context.Background(), accessToken) + ti, err := p.oauthServer.LoadAccessToken(ctx, accessToken) if err != nil { return nil, fmt.Errorf("AuthorizeStreamingRequest: error loading access token: %s", err) } diff --git a/internal/processing/streaming/authorize_test.go b/internal/processing/streaming/authorize_test.go new file mode 100644 index 000000000..f52396250 --- /dev/null +++ b/internal/processing/streaming/authorize_test.go @@ -0,0 +1,48 @@ +/* + GoToSocial + Copyright (C) 2021 GoToSocial Authors admin@gotosocial.org + + This program is free software: you can redistribute it and/or modify + it under the terms of the GNU Affero General Public License as published by + the Free Software Foundation, either version 3 of the License, or + (at your option) any later version. + + This program is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU Affero General Public License for more details. + + You should have received a copy of the GNU Affero General Public License + along with this program. If not, see <http://www.gnu.org/licenses/>. +*/ + +package streaming_test + +import ( + "context" + "testing" + + "github.com/stretchr/testify/suite" +) + +type AuthorizeTestSuite struct { + StreamingTestSuite +} + +func (suite *AuthorizeTestSuite) TestAuthorize() { + account1, err := suite.streamingProcessor.AuthorizeStreamingRequest(context.Background(), suite.testTokens["local_account_1"].Access) + suite.NoError(err) + suite.Equal(suite.testAccounts["local_account_1"].ID, account1.ID) + + account2, err := suite.streamingProcessor.AuthorizeStreamingRequest(context.Background(), suite.testTokens["local_account_2"].Access) + suite.NoError(err) + suite.Equal(suite.testAccounts["local_account_2"].ID, account2.ID) + + noAccount, err := suite.streamingProcessor.AuthorizeStreamingRequest(context.Background(), "aaaaaaaaaaaaaaaaaaaaa!!") + suite.EqualError(err, "AuthorizeStreamingRequest: error loading access token: no entries") + suite.Nil(noAccount) +} + +func TestAuthorizeTestSuite(t *testing.T) { + suite.Run(t, &AuthorizeTestSuite{}) +} diff --git a/internal/processing/streaming/notification.go b/internal/processing/streaming/notification.go new file mode 100644 index 000000000..870490be4 --- /dev/null +++ b/internal/processing/streaming/notification.go @@ -0,0 +1,37 @@ +/* + GoToSocial + Copyright (C) 2021 GoToSocial Authors admin@gotosocial.org + + This program is free software: you can redistribute it and/or modify + it under the terms of the GNU Affero General Public License as published by + the Free Software Foundation, either version 3 of the License, or + (at your option) any later version. + + This program is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU Affero General Public License for more details. + + You should have received a copy of the GNU Affero General Public License + along with this program. If not, see <http://www.gnu.org/licenses/>. +*/ + +package streaming + +import ( + "encoding/json" + "fmt" + + apimodel "github.com/superseriousbusiness/gotosocial/internal/api/model" + "github.com/superseriousbusiness/gotosocial/internal/gtsmodel" + "github.com/superseriousbusiness/gotosocial/internal/stream" +) + +func (p *processor) StreamNotificationToAccount(n *apimodel.Notification, account *gtsmodel.Account) error { + bytes, err := json.Marshal(n) + if err != nil { + return fmt.Errorf("error marshalling notification to json: %s", err) + } + + return p.streamToAccount(string(bytes), stream.EventTypeNotification, account.ID) +} diff --git a/internal/processing/streaming/notification_test.go b/internal/processing/streaming/notification_test.go new file mode 100644 index 000000000..fa77a8f92 --- /dev/null +++ b/internal/processing/streaming/notification_test.go @@ -0,0 +1,60 @@ +/* + GoToSocial + Copyright (C) 2021 GoToSocial Authors admin@gotosocial.org + + This program is free software: you can redistribute it and/or modify + it under the terms of the GNU Affero General Public License as published by + the Free Software Foundation, either version 3 of the License, or + (at your option) any later version. + + This program is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU Affero General Public License for more details. + + You should have received a copy of the GNU Affero General Public License + along with this program. If not, see <http://www.gnu.org/licenses/>. +*/ + +package streaming_test + +import ( + "context" + "testing" + + "github.com/stretchr/testify/suite" + apimodel "github.com/superseriousbusiness/gotosocial/internal/api/model" + "github.com/superseriousbusiness/gotosocial/testrig" +) + +type NotificationTestSuite struct { + StreamingTestSuite +} + +func (suite *NotificationTestSuite) TestStreamNotification() { + account := suite.testAccounts["local_account_1"] + + openStream, errWithCode := suite.streamingProcessor.OpenStreamForAccount(context.Background(), account, "user") + suite.NoError(errWithCode) + + followAccount := suite.testAccounts["remote_account_1"] + followAccountAPIModel, err := testrig.NewTestTypeConverter(suite.db).AccountToAPIAccountPublic(context.Background(), followAccount) + suite.NoError(err) + + notification := &apimodel.Notification{ + ID: "01FH57SJCMDWQGEAJ0X08CE3WV", + Type: "follow", + CreatedAt: "2021-10-04T10:52:36+02:00", + Account: followAccountAPIModel, + } + + err = suite.streamingProcessor.StreamNotificationToAccount(notification, account) + suite.NoError(err) + + msg := <-openStream.Messages + suite.Equal(`{"id":"01FH57SJCMDWQGEAJ0X08CE3WV","type":"follow","created_at":"2021-10-04T10:52:36+02:00","account":{"id":"01F8MH5ZK5VRH73AKHQM6Y9VNX","username":"foss_satan","acct":"foss_satan@fossbros-anonymous.io","display_name":"big gerald","locked":false,"bot":false,"created_at":"2021-09-26T12:52:36+02:00","note":"i post about like, i dunno, stuff, or whatever!!!!","url":"http://fossbros-anonymous.io/@foss_satan","avatar":"","avatar_static":"","header":"","header_static":"","followers_count":0,"following_count":0,"statuses_count":0,"last_status_at":"","emojis":[],"fields":[]}}`, msg.Payload) +} + +func TestNotificationTestSuite(t *testing.T) { + suite.Run(t, &NotificationTestSuite{}) +} diff --git a/internal/processing/streaming/openstream.go b/internal/processing/streaming/openstream.go index d4e4eef9f..74b6486f5 100644 --- a/internal/processing/streaming/openstream.go +++ b/internal/processing/streaming/openstream.go @@ -1,3 +1,21 @@ +/* + GoToSocial + Copyright (C) 2021 GoToSocial Authors admin@gotosocial.org + + This program is free software: you can redistribute it and/or modify + it under the terms of the GNU Affero General Public License as published by + the Free Software Foundation, either version 3 of the License, or + (at your option) any later version. + + This program is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU Affero General Public License for more details. + + You should have received a copy of the GNU Affero General Public License + along with this program. If not, see <http://www.gnu.org/licenses/>. +*/ + package streaming import ( diff --git a/internal/processing/streaming/openstream_test.go b/internal/processing/streaming/openstream_test.go new file mode 100644 index 000000000..6c4134997 --- /dev/null +++ b/internal/processing/streaming/openstream_test.go @@ -0,0 +1,41 @@ +/* + GoToSocial + Copyright (C) 2021 GoToSocial Authors admin@gotosocial.org + + This program is free software: you can redistribute it and/or modify + it under the terms of the GNU Affero General Public License as published by + the Free Software Foundation, either version 3 of the License, or + (at your option) any later version. + + This program is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU Affero General Public License for more details. + + You should have received a copy of the GNU Affero General Public License + along with this program. If not, see <http://www.gnu.org/licenses/>. +*/ + +package streaming_test + +import ( + "context" + "testing" + + "github.com/stretchr/testify/suite" +) + +type OpenStreamTestSuite struct { + StreamingTestSuite +} + +func (suite *OpenStreamTestSuite) TestOpenStream() { + account := suite.testAccounts["local_account_1"] + + _, errWithCode := suite.streamingProcessor.OpenStreamForAccount(context.Background(), account, "user") + suite.NoError(errWithCode) +} + +func TestOpenStreamTestSuite(t *testing.T) { + suite.Run(t, &OpenStreamTestSuite{}) +} diff --git a/internal/processing/streaming/streamdelete.go b/internal/processing/streaming/streamdelete.go index cd541bc57..8332c37dc 100644 --- a/internal/processing/streaming/streamdelete.go +++ b/internal/processing/streaming/streamdelete.go @@ -1,3 +1,21 @@ +/* + GoToSocial + Copyright (C) 2021 GoToSocial Authors admin@gotosocial.org + + This program is free software: you can redistribute it and/or modify + it under the terms of the GNU Affero General Public License as published by + the Free Software Foundation, either version 3 of the License, or + (at your option) any later version. + + This program is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU Affero General Public License for more details. + + You should have received a copy of the GNU Affero General Public License + along with this program. If not, see <http://www.gnu.org/licenses/>. +*/ + package streaming import ( @@ -10,39 +28,20 @@ import ( func (p *processor) StreamDelete(statusID string) error { errs := []string{} - // we want to range through ALL streams for ALL accounts here to make sure it's very clear to everyone that the status has been deleted - p.streamMap.Range(func(k interface{}, v interface{}) bool { - // the key of this map should be an accountID (string) - accountID, ok := k.(string) - if !ok { - errs = append(errs, "key in streamMap was not a string!") - return false - } - - // the value of the map should be a buncha streams - streamsForAccount, ok := v.(*stream.StreamsForAccount) - if !ok { - errs = append(errs, fmt.Sprintf("stream map error for account stream %s", accountID)) - } - - // lock the streams while we work on them - streamsForAccount.Lock() - defer streamsForAccount.Unlock() - for _, s := range streamsForAccount.Streams { - // lock each individual stream as we work on it - s.Lock() - defer s.Unlock() - if s.Connected { - s.Messages <- &stream.Message{ - Stream: []string{s.Type}, - Event: "delete", - Payload: statusID, - } - } - } + // get all account IDs with open streams + accountIDs := []string{} + p.streamMap.Range(func(k interface{}, _ interface{}) bool { + accountIDs = append(accountIDs, k.(string)) return true }) + // stream the delete to every account + for _, accountID := range accountIDs { + if err := p.streamToAccount(statusID, stream.EventTypeDelete, accountID); err != nil { + errs = append(errs, err.Error()) + } + } + if len(errs) != 0 { return fmt.Errorf("one or more errors streaming status delete: %s", strings.Join(errs, ";")) } diff --git a/internal/processing/streaming/streaming.go b/internal/processing/streaming/streaming.go index 610d4a9d2..abce30cd1 100644 --- a/internal/processing/streaming/streaming.go +++ b/internal/processing/streaming/streaming.go @@ -1,3 +1,21 @@ +/* + GoToSocial + Copyright (C) 2021 GoToSocial Authors admin@gotosocial.org + + This program is free software: you can redistribute it and/or modify + it under the terms of the GNU Affero General Public License as published by + the Free Software Foundation, either version 3 of the License, or + (at your option) any later version. + + This program is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU Affero General Public License for more details. + + You should have received a copy of the GNU Affero General Public License + along with this program. If not, see <http://www.gnu.org/licenses/>. +*/ + package streaming import ( @@ -6,14 +24,11 @@ import ( "github.com/sirupsen/logrus" apimodel "github.com/superseriousbusiness/gotosocial/internal/api/model" - "github.com/superseriousbusiness/gotosocial/internal/config" "github.com/superseriousbusiness/gotosocial/internal/db" "github.com/superseriousbusiness/gotosocial/internal/gtserror" "github.com/superseriousbusiness/gotosocial/internal/gtsmodel" "github.com/superseriousbusiness/gotosocial/internal/oauth" "github.com/superseriousbusiness/gotosocial/internal/stream" - "github.com/superseriousbusiness/gotosocial/internal/typeutils" - "github.com/superseriousbusiness/gotosocial/internal/visibility" ) // Processor wraps a bunch of functions for processing streaming. @@ -22,8 +37,8 @@ type Processor interface { AuthorizeStreamingRequest(ctx context.Context, accessToken string) (*gtsmodel.Account, error) // OpenStreamForAccount returns a new Stream for the given account, which will contain a channel for passing messages back to the caller. OpenStreamForAccount(ctx context.Context, account *gtsmodel.Account, streamType string) (*stream.Stream, gtserror.WithCode) - // StreamStatusToAccount streams the given status to any open, appropriate streams belonging to the given account. - StreamStatusToAccount(s *apimodel.Status, account *gtsmodel.Account) error + // StreamUpdateToAccount streams the given update to any open, appropriate streams belonging to the given account. + StreamUpdateToAccount(s *apimodel.Status, account *gtsmodel.Account) error // StreamNotificationToAccount streams the given notification to any open, appropriate streams belonging to the given account. StreamNotificationToAccount(n *apimodel.Notification, account *gtsmodel.Account) error // StreamDelete streams the delete of the given statusID to *ALL* open streams. @@ -31,22 +46,16 @@ type Processor interface { } type processor struct { - tc typeutils.TypeConverter - config *config.Config db db.DB - filter visibility.Filter log *logrus.Logger oauthServer oauth.Server streamMap *sync.Map } // New returns a new status processor. -func New(db db.DB, tc typeutils.TypeConverter, oauthServer oauth.Server, config *config.Config, log *logrus.Logger) Processor { +func New(db db.DB, oauthServer oauth.Server, log *logrus.Logger) Processor { return &processor{ - tc: tc, - config: config, db: db, - filter: visibility.NewFilter(db, log), log: log, oauthServer: oauthServer, streamMap: &sync.Map{}, diff --git a/internal/processing/streaming/streaming_test.go b/internal/processing/streaming/streaming_test.go new file mode 100644 index 000000000..acc090b06 --- /dev/null +++ b/internal/processing/streaming/streaming_test.go @@ -0,0 +1,55 @@ +/* + GoToSocial + Copyright (C) 2021 GoToSocial Authors admin@gotosocial.org + + This program is free software: you can redistribute it and/or modify + it under the terms of the GNU Affero General Public License as published by + the Free Software Foundation, either version 3 of the License, or + (at your option) any later version. + + This program is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU Affero General Public License for more details. + + You should have received a copy of the GNU Affero General Public License + along with this program. If not, see <http://www.gnu.org/licenses/>. +*/ + +package streaming_test + +import ( + "github.com/sirupsen/logrus" + "github.com/stretchr/testify/suite" + "github.com/superseriousbusiness/gotosocial/internal/db" + "github.com/superseriousbusiness/gotosocial/internal/gtsmodel" + "github.com/superseriousbusiness/gotosocial/internal/oauth" + "github.com/superseriousbusiness/gotosocial/internal/processing/streaming" + "github.com/superseriousbusiness/gotosocial/testrig" +) + +type StreamingTestSuite struct { + suite.Suite + testAccounts map[string]*gtsmodel.Account + testTokens map[string]*gtsmodel.Token + db db.DB + oauthServer oauth.Server + log *logrus.Logger + + streamingProcessor streaming.Processor +} + +func (suite *StreamingTestSuite) SetupTest() { + suite.testAccounts = testrig.NewTestAccounts() + suite.testTokens = testrig.NewTestTokens() + suite.db = testrig.NewTestDB() + suite.oauthServer = testrig.NewTestOauthServer(suite.db) + suite.log = testrig.NewTestLog() + suite.streamingProcessor = streaming.New(suite.db, suite.oauthServer, suite.log) + + testrig.StandardDBSetup(suite.db, suite.testAccounts) +} + +func (suite *StreamingTestSuite) TearDownTest() { + testrig.StandardDBTeardown(suite.db) +} diff --git a/internal/processing/streaming/streamnotification.go b/internal/processing/streaming/streamnotification.go deleted file mode 100644 index d8460874f..000000000 --- a/internal/processing/streaming/streamnotification.go +++ /dev/null @@ -1,51 +0,0 @@ -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" - "github.com/superseriousbusiness/gotosocial/internal/stream" -) - -func (p *processor) StreamNotificationToAccount(n *apimodel.Notification, account *gtsmodel.Account) error { - l := p.log.WithFields(logrus.Fields{ - "func": "StreamNotificationToAccount", - "account": account.ID, - }) - v, ok := p.streamMap.Load(account.ID) - if !ok { - // no open connections so nothing to stream - return nil - } - - streamsForAccount, ok := v.(*stream.StreamsForAccount) - if !ok { - return errors.New("stream map error") - } - - notificationBytes, err := json.Marshal(n) - if err != nil { - return fmt.Errorf("error marshalling notification to json: %s", err) - } - - streamsForAccount.Lock() - defer streamsForAccount.Unlock() - for _, s := range streamsForAccount.Streams { - s.Lock() - defer s.Unlock() - if s.Connected { - l.Debugf("streaming notification to stream id %s", s.ID) - s.Messages <- &stream.Message{ - Stream: []string{s.Type}, - Event: "notification", - Payload: string(notificationBytes), - } - } - } - - return nil -} diff --git a/internal/processing/streaming/streamstatus.go b/internal/processing/streaming/streamstatus.go deleted file mode 100644 index f4d6b2629..000000000 --- a/internal/processing/streaming/streamstatus.go +++ /dev/null @@ -1,51 +0,0 @@ -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" - "github.com/superseriousbusiness/gotosocial/internal/stream" -) - -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.(*stream.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 _, s := range streamsForAccount.Streams { - s.Lock() - defer s.Unlock() - if s.Connected { - l.Debugf("streaming status to stream id %s", s.ID) - s.Messages <- &stream.Message{ - Stream: []string{s.Type}, - Event: "update", - Payload: string(statusBytes), - } - } - } - - return nil -} diff --git a/internal/processing/streaming/streamtoaccount.go b/internal/processing/streaming/streamtoaccount.go new file mode 100644 index 000000000..140910ab7 --- /dev/null +++ b/internal/processing/streaming/streamtoaccount.go @@ -0,0 +1,55 @@ +/* + GoToSocial + Copyright (C) 2021 GoToSocial Authors admin@gotosocial.org + + This program is free software: you can redistribute it and/or modify + it under the terms of the GNU Affero General Public License as published by + the Free Software Foundation, either version 3 of the License, or + (at your option) any later version. + + This program is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU Affero General Public License for more details. + + You should have received a copy of the GNU Affero General Public License + along with this program. If not, see <http://www.gnu.org/licenses/>. +*/ + +package streaming + +import ( + "errors" + + "github.com/superseriousbusiness/gotosocial/internal/stream" +) + +// streamToAccount streams the given payload with the given event type to any streams currently open for the given account ID. +func (p *processor) streamToAccount(payload string, event stream.EventType, accountID string) error { + v, ok := p.streamMap.Load(accountID) + if !ok { + // no open connections so nothing to stream + return nil + } + + streamsForAccount, ok := v.(*stream.StreamsForAccount) + if !ok { + return errors.New("stream map error") + } + + streamsForAccount.Lock() + defer streamsForAccount.Unlock() + for _, s := range streamsForAccount.Streams { + s.Lock() + defer s.Unlock() + if s.Connected { + s.Messages <- &stream.Message{ + Stream: []string{s.Type}, + Event: string(event), + Payload: payload, + } + } + } + + return nil +} diff --git a/internal/processing/streaming/update.go b/internal/processing/streaming/update.go new file mode 100644 index 000000000..da7dcb6ce --- /dev/null +++ b/internal/processing/streaming/update.go @@ -0,0 +1,37 @@ +/* + GoToSocial + Copyright (C) 2021 GoToSocial Authors admin@gotosocial.org + + This program is free software: you can redistribute it and/or modify + it under the terms of the GNU Affero General Public License as published by + the Free Software Foundation, either version 3 of the License, or + (at your option) any later version. + + This program is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU Affero General Public License for more details. + + You should have received a copy of the GNU Affero General Public License + along with this program. If not, see <http://www.gnu.org/licenses/>. +*/ + +package streaming + +import ( + "encoding/json" + "fmt" + + apimodel "github.com/superseriousbusiness/gotosocial/internal/api/model" + "github.com/superseriousbusiness/gotosocial/internal/gtsmodel" + "github.com/superseriousbusiness/gotosocial/internal/stream" +) + +func (p *processor) StreamUpdateToAccount(s *apimodel.Status, account *gtsmodel.Account) error { + bytes, err := json.Marshal(s) + if err != nil { + return fmt.Errorf("error marshalling status to json: %s", err) + } + + return p.streamToAccount(string(bytes), stream.EventTypeUpdate, account.ID) +} |