start messing about with streaming api

This commit is contained in:
tsmethurst 2021-06-17 19:20:08 +02:00
commit 0cee5aa569
9 changed files with 241 additions and 2 deletions

View file

@ -0,0 +1,33 @@
package streaming
import (
"context"
"fmt"
"github.com/superseriousbusiness/gotosocial/internal/gtsmodel"
)
func (p *processor) AuthorizeStreamingRequest(accessToken string) (*gtsmodel.Account, error) {
ti, err := p.oauthServer.LoadAccessToken(context.Background(), accessToken)
if err != nil {
return nil, fmt.Errorf("AuthorizeStreamingRequest: error loading access token: %s", err)
}
uid := ti.GetUserID()
if uid == "" {
return nil, fmt.Errorf("AuthorizeStreamingRequest: no userid in token")
}
// fetch user's and account for this user id
user := &gtsmodel.User{}
if err := p.db.GetByID(uid, user); err != nil || user == nil {
return nil, fmt.Errorf("AuthorizeStreamingRequest: no user found for validated uid %s", uid)
}
acct := &gtsmodel.Account{}
if err := p.db.GetByID(user.AccountID, acct); err != nil || acct == nil {
return nil, fmt.Errorf("AuthorizeStreamingRequest: no account retrieved for user with id %s", uid)
}
return acct, nil
}

View file

@ -0,0 +1,22 @@
package streaming
import (
"github.com/gorilla/websocket"
"github.com/superseriousbusiness/gotosocial/internal/gtserror"
"github.com/superseriousbusiness/gotosocial/internal/gtsmodel"
)
func (p *processor) StreamForAccount(c *websocket.Conn, account *gtsmodel.Account, streamType string) gtserror.WithCode {
v, loaded := p.streamMap.LoadOrStore(account.ID, sync.Slice)
if loaded {
}
return nil
}
type streams struct {
accountID string
}

View file

@ -0,0 +1,45 @@
package streaming
import (
"sync"
"github.com/gorilla/websocket"
"github.com/sirupsen/logrus"
"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/typeutils"
"github.com/superseriousbusiness/gotosocial/internal/visibility"
)
// Processor wraps a bunch of functions for processing streaming.
type Processor interface {
// AuthorizeStreamingRequest returns an oauth2 token info in response to an access token query from the streaming API
AuthorizeStreamingRequest(accessToken string) (*gtsmodel.Account, error)
StreamForAccount(c *websocket.Conn, account *gtsmodel.Account, streamType string) gtserror.WithCode
}
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 {
return &processor{
tc: tc,
config: config,
db: db,
filter: visibility.NewFilter(db, log),
log: log,
oauthServer: oauthServer,
streamMap: &sync.Map{},
}
}