[feature] Support multiple subscriptions on single websocket connection (#1489)

- Allow Oauth authentication on websocket endpoint
- Make streamType query parameter optional
- Read websocket commands from client and update subscriptions

2 files changed, 13 insertions, 2 deletions

diff --git a/internal/processing/stream/open.go b/internal/processing/stream/open.go
index 10d01a767..823efa182 100644
--- a/internal/processing/stream/open.go
+++ b/internal/processing/stream/open.go
@@ -45,9 +45,17 @@ func (p *Processor) Open(ctx context.Context, account *gtsmodel.Account, streamT
 		return nil, gtserror.NewErrorInternalError(fmt.Errorf("error generating stream id: %s", err))
 	}
 
+	// Each stream can be subscibed to multiple timelines.
+	// 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
+	}
+
 	thisStream := &stream.Stream{
 		ID:        streamID,
-		Timeline:  streamTimeline,
+		Timelines: timelines,
 		Messages:  make(chan *stream.Message, 100),
 		Hangup:    make(chan interface{}, 1),
 		Connected: true,
diff --git a/internal/processing/stream/stream.go b/internal/processing/stream/stream.go
index a10ab2474..f3a9e92f3 100644
--- a/internal/processing/stream/stream.go
+++ b/internal/processing/stream/stream.go
@@ -63,12 +63,15 @@ func (p *Processor) toAccount(payload string, event string, timelines []string,
 		}
 
 		for _, t := range timelines {
-			if s.Timeline == string(t) {
+			if _, found := s.Timelines[t]; found {
 				s.Messages <- &stream.Message{
 					Stream:  []string{string(t)},
 					Event:   string(event),
 					Payload: payload,
 				}
+				// break out to the outer loop, to avoid sending duplicates
+				// of the same event to the same stream
+				break
 			}
 		}
 	}