diff options
author | 2023-05-25 10:37:38 +0200 | |
---|---|---|
committer | 2023-05-25 10:37:38 +0200 | |
commit | f5c004d67d4ed66b6c6df100afec47174aa14ae0 (patch) | |
tree | 45b72a6e90450d711e10571d844138186fe023c9 /internal/processing/stream/open.go | |
parent | [docs] local docs hacking howto (#1816) (diff) | |
download | gotosocial-f5c004d67d4ed66b6c6df100afec47174aa14ae0.tar.xz |
[feature] Add List functionality (#1802)
* start working on lists
* further list work
* test list db functions nicely
* more work on lists
* peepoopeepoo
* poke
* start list timeline func
* we're getting there lads
* couldn't be me working on stuff... could it?
* hook up handlers
* fiddling
* weeee
* woah
* screaming, pissing
* fix streaming being a whiny baby
* lint, small test fix, swagger
* tidying up, testing
* fucked! by the linter
* move timelines to state like a boss
* add timeline start to tests using state
* invalidate lists
Diffstat (limited to 'internal/processing/stream/open.go')
-rw-r--r-- | internal/processing/stream/open.go | 67 |
1 files changed, 36 insertions, 31 deletions
diff --git a/internal/processing/stream/open.go b/internal/processing/stream/open.go index e43152b29..1c041309f 100644 --- a/internal/processing/stream/open.go +++ b/internal/processing/stream/open.go @@ -31,60 +31,65 @@ import ( ) // Open returns a new Stream for the given account, which will contain a channel for passing messages back to the caller. -func (p *Processor) Open(ctx context.Context, account *gtsmodel.Account, streamTimeline string) (*stream.Stream, gtserror.WithCode) { +func (p *Processor) Open(ctx context.Context, account *gtsmodel.Account, streamType string) (*stream.Stream, gtserror.WithCode) { l := log.WithContext(ctx).WithFields(kv.Fields{ {"account", account.ID}, - {"streamType", streamTimeline}, + {"streamType", streamType}, }...) l.Debug("received open stream request") - // each stream needs a unique ID so we know to close it - streamID, err := id.NewRandomULID() + var ( + streamID string + err error + ) + + // Each stream needs a unique ID so we know to close it. + streamID, err = id.NewRandomULID() if err != nil { - return nil, gtserror.NewErrorInternalError(fmt.Errorf("error generating stream id: %s", err)) + return nil, gtserror.NewErrorInternalError(fmt.Errorf("error generating stream id: %w", err)) } - // Each stream can be subscibed to multiple timelines. + // Each stream can be subscibed to multiple types. // Record them in a set, and include the initial one - // if it was given to us - timelines := map[string]bool{} - if streamTimeline != "" { - timelines[streamTimeline] = true + // if it was given to us. + streamTypes := map[string]any{} + if streamType != "" { + streamTypes[streamType] = true } - thisStream := &stream.Stream{ - ID: streamID, - Timelines: timelines, - Messages: make(chan *stream.Message, 100), - Hangup: make(chan interface{}, 1), - Connected: true, + newStream := &stream.Stream{ + ID: streamID, + StreamTypes: streamTypes, + Messages: make(chan *stream.Message, 100), + Hangup: make(chan interface{}, 1), + Connected: true, } - go p.waitToCloseStream(account, thisStream) + go p.waitToCloseStream(account, newStream) v, ok := p.streamMap.Load(account.ID) - if !ok || v == nil { - // there is no entry in the streamMap for this account yet, so make one and store it - streamsForAccount := &stream.StreamsForAccount{ - Streams: []*stream.Stream{ - thisStream, - }, - } - p.streamMap.Store(account.ID, streamsForAccount) - } else { - // there is an entry in the streamMap for this account - // parse the interface as a streamsForAccount + if ok { + // There is an entry in the streamMap + // for this account. Parse it out. streamsForAccount, ok := v.(*stream.StreamsForAccount) if !ok { return nil, gtserror.NewErrorInternalError(errors.New("stream map error")) } - // append this stream to it + // Append new stream to existing entry. streamsForAccount.Lock() - streamsForAccount.Streams = append(streamsForAccount.Streams, thisStream) + streamsForAccount.Streams = append(streamsForAccount.Streams, newStream) streamsForAccount.Unlock() + } else { + // There is no entry in the streamMap for + // this account yet. Create one and store it. + p.streamMap.Store(account.ID, &stream.StreamsForAccount{ + Streams: []*stream.Stream{ + newStream, + }, + }) } - return thisStream, nil + return newStream, nil } // waitToCloseStream waits until the hangup channel is closed for the given stream. |