Something went wrong. Try again.
An atproto PDS written in Go
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140package server
import ( "context" "strconv" "time"
"github.com/bluesky-social/indigo/events" "github.com/bluesky-social/indigo/lex/util" "github.com/btcsuite/websocket" "github.com/haileyok/cocoon/metrics" "github.com/labstack/echo/v4")
func (s *Server) handleSyncSubscribeRepos(e echo.Context) error { ctx, cancel := context.WithCancel(e.Request().Context()) defer cancel()
logger := s.logger.With("component", "subscribe-repos-websocket")
conn, err := websocket.Upgrade(e.Response().Writer, e.Request(), e.Response().Header(), 1<<10, 1<<10) if err != nil { logger.Error("unable to establish websocket with relay", "err", err) return err }
ident := e.RealIP() + "-" + e.Request().UserAgent() logger = logger.With("ident", ident) logger.Info("new connection established")
var since *int64 if cursorStr := e.QueryParam("cursor"); cursorStr != "" { cursor, err := strconv.ParseInt(cursorStr, 10, 64) if err != nil { logger.Warn("invalid cursor parameter", "cursor", cursorStr, "err", err) } else { since = &cursor logger.Info("subscribing with cursor", "cursor", cursor) } }
metrics.RelaysConnected.WithLabelValues(ident).Inc() defer func() { metrics.RelaysConnected.WithLabelValues(ident).Dec() }()
evts, evtManCancel, err := s.evtman.Subscribe(ctx, ident, func(evt *events.XRPCStreamEvent) bool { return true }, since) if err != nil { return err } defer evtManCancel()
// drop the connection whenever a subscriber disconnects from the socket, we should get errors go func() { for { select { case <-ctx.Done(): return default: if _, _, err := conn.ReadMessage(); err != nil { logger.Warn("websocket error", "err", err) cancel() return } } } }()
header := events.EventHeader{Op: events.EvtKindMessage} for evt := range evts { func() { defer func() { metrics.RelaySends.WithLabelValues(ident, header.MsgType).Inc() }()
wc, err := conn.NextWriter(websocket.BinaryMessage) if err != nil { logger.Error("error writing message to relay", "err", err) return }
if ctx.Err() != nil { logger.Error("context error", "err", err) return }
var obj util.CBOR switch { case evt.Error != nil: header.Op = events.EvtKindErrorFrame obj = evt.Error case evt.RepoCommit != nil: header.MsgType = "#commit" obj = evt.RepoCommit case evt.RepoIdentity != nil: header.MsgType = "#identity" obj = evt.RepoIdentity case evt.RepoAccount != nil: header.MsgType = "#account" obj = evt.RepoAccount case evt.RepoInfo != nil: header.MsgType = "#info" obj = evt.RepoInfo default: logger.Warn("unrecognized event kind") return }
if err := header.MarshalCBOR(wc); err != nil { logger.Error("failed to write header to relay", "err", err) return }
if err := obj.MarshalCBOR(wc); err != nil { logger.Error("failed to write event to relay", "err", err) return }
if err := wc.Close(); err != nil { logger.Error("failed to flush-close our event write", "err", err) return } }() }
// we should tell the relay to request a new crawl at this point if we got disconnected // use a new context since the old one might be cancelled at this point go func() { retryCtx, retryCancel := context.WithTimeout(context.Background(), 10*time.Second) defer retryCancel() if err := s.requestCrawl(retryCtx); err != nil { logger.Error("error requesting crawls", "err", err) } }()
return nil}