sessions

type Closer ¶

type Closer interface {
	Close(reason domain.SessionCloseReason, addBytes int64)
	Touch(addBytes int64)
}

Closer ends a connection-bound session. addBytes is the cumulative bytes sent on the underlying transport; pass 0 if the protocol can't measure. Both forms are idempotent.

Touch refreshes the session's UpdatedAt so the idle reaper (which closes any session whose UpdatedAt is older than `idle_timeout_sec`, default 30s) does not mistake an actively-streaming RTMP / SRT / RTSP session for an abandoned one. Connection-bound protocols don't naturally update UpdatedAt the way HLS does (every segment GET refreshes it), so the publisher must call Touch from its data-write hot path. Throttling is the caller's responsibility — see playSession.touch.

addBytes credits an incremental byte delta to the live session record so the API surfaces a growing counter mid-stream (RTMP / SRT / MPEGTS); pass 0 when the protocol can only measure the total at close time (RTSP — gortsplib serialises packets internally).

type ConnHit ¶

type ConnHit struct {
	StreamCode domain.StreamCode
	Protocol   domain.SessionProto // SessionProtoRTMP / SessionProtoSRT / SessionProtoRTSP
	RemoteAddr string              // ip:port from net.Conn.RemoteAddr().String()
	UserAgent  string              // RTMP flashVer / RTSP User-Agent / "" for SRT
	Token      string              // optional, e.g. SRT streamid query
	Secure     bool                // true for RTMPS / SRTS / RTSPS
}

ConnHit is the input to OpenConn for connection-bound protocols.

type Filter ¶

type Filter struct {
	StreamCode domain.StreamCode // exact match; "" = any
	Protocol   domain.SessionProto
	Status     FilterStatus
	Limit      int // <=0 = unlimited
}

Filter is the union of selectors accepted by List. Zero value matches every session.

type FilterStatus ¶

type FilterStatus string

FilterStatus narrows List by lifecycle state.

const (
	FilterStatusAny    FilterStatus = ""       // active + recently closed
	FilterStatusActive FilterStatus = "active" // ClosedAt == nil
	FilterStatusClosed FilterStatus = "closed" // ClosedAt != nil
)

type GeoIPResolver ¶

type GeoIPResolver interface {
	Country(ip net.IP) string
}

GeoIPResolver maps a remote IP to an ISO 3166-1 alpha-2 country code. Implementations must be safe for concurrent use and return "" when the address cannot be resolved (private address, missing DB, lookup error).

The default registered in DI is NullGeoIP — no lookup, Country always "". When `sessions.geoip_db_path` is set in config, the wiring layer in `cmd/server/main.go` opens the .mmdb via NewMaxMindGeoIP and registers that as the resolver instead. Operators can also swap in any custom implementation (IP2Location, in-house service) by replacing the DI binding before service start.

type HTTPHit ¶

type HTTPHit struct {
	StreamCode domain.StreamCode
	Protocol   domain.SessionProto // SessionProtoHLS or SessionProtoDASH
	IP         string              // already extracted; no port
	UserAgent  string
	Referer    string
	Query      string // raw r.URL.RawQuery — stored only on session open
	Token      string // ?token=… value if present
	Secure     bool
	BytesDelta int64 // bytes the most recent response wrote to the client
}

HTTPHit is the input to TrackHTTP. Build with NewHTTPHit from an http.Request so the IP / UA / Referer extraction is consistent.

func NewHTTPHit(r *http.Request, code domain.StreamCode, proto domain.SessionProto, bytesDelta int64) HTTPHit

type MaxMindGeoIP ¶ added in v0.0.92

type MaxMindGeoIP struct {
	// contains filtered or unexported fields
}

MaxMindGeoIP wraps a MaxMind .mmdb reader and resolves IPs into ISO 3166-1 alpha-2 country codes for PlaySession.Country.

func NewMaxMindGeoIP(path string) (*MaxMindGeoIP, error)

func (m *MaxMindGeoIP) Close() error

func (m *MaxMindGeoIP) Country(ip net.IP) string

type NullGeoIP ¶

type NullGeoIP struct{}

NullGeoIP is a GeoIPResolver that always returns "". Used when GeoIP is disabled or no real resolver has been wired. Public so external code can reference it as an explicit "disabled" marker.

func (NullGeoIP) Country(net.IP) string

type Service ¶

type Service struct {
	// contains filtered or unexported fields
}

Service is the public DI handle. It implements Tracker.

func New(i do.Injector) (*Service, error)

func NewServiceForTesting(cfg config.SessionsConfig, bus events.Bus, geo GeoIPResolver) *Service

func (s Service) Get(id string) (*domain.PlaySession, bool)

func (s Service) Kick(id string) bool

func (s Service) List(filter Filter) []*domain.PlaySession

func (s Service) OpenConn(ctx context.Context, h ConnHit) (*domain.PlaySession, Closer)

func (s *Service) Run(ctx context.Context)

func (s Service) Stats() Stats

func (s Service) TrackHTTP(ctx context.Context, h HTTPHit) *domain.PlaySession

func (s *Service) UpdateConfig(cfg config.SessionsConfig)

type Stats ¶

type Stats struct {
	Active          int   `json:"active"`
	OpenedTotal     int64 `json:"opened_total"`
	ClosedTotal     int64 `json:"closed_total"`
	IdleClosedTotal int64 `json:"idle_closed_total"`
	KickedTotal     int64 `json:"kicked_total"`
}

Stats summarises the live tracker state.

type Tracker ¶

type Tracker interface {
	// TrackHTTP records activity for an HTTP-based pull protocol (HLS / DASH).
	// Called once per segment / manifest GET. Idempotent on the (stream, fp)
	// pair within the idle window — repeated calls extend UpdatedAt and
	// accumulate Bytes onto an existing session instead of opening a new one.
	//
	// The returned session is a snapshot copy safe to retain by the caller;
	// further mutations happen on the live record held inside the tracker.
	TrackHTTP(ctx context.Context, h HTTPHit) *domain.PlaySession

	// OpenConn opens a session for a connection-bound protocol (RTMP / SRT /
	// RTSP). Returns the session record and a Closer the caller must invoke
	// when the underlying transport ends. The Closer is idempotent.
	OpenConn(ctx context.Context, h ConnHit) (*domain.PlaySession, Closer)

	// List returns a snapshot of every session matching the filter. Active
	// (ClosedAt == nil) and idle entries are included; filter by Status.
	List(filter Filter) []*domain.PlaySession

	// Get returns the session with the given ID, or false when missing.
	Get(id string) (*domain.PlaySession, bool)

	// Kick force-closes an active session and emits a closed event with
	// reason="kicked". Returns false when the id is unknown or already closed.
	Kick(id string) bool

	// Stats returns aggregate counters useful for /metrics or readiness
	// probes. Does not allocate per-session storage.
	Stats() Stats
}

Tracker is the public contract that publisher / api layers use to record playback sessions. The implementation is in-memory; restart drops state.

All methods are safe for concurrent use.