mirror of
https://github.com/truewhile/MeBox.git
synced 2026-10-03 12:26:36 +08:00
feat: bootstrap MediaStationGo (Go + React rewrite of MediaStation)
Adopt the cropflre/nowen-video tech stack and rebuild the project from
scratch:
- Backend: Go 1.25 + Gin + GORM + SQLite (WAL) + JWT + WebSocket hub.
Layered packages config / database / model / repository / service /
middleware / handler. Default admin (admin/admin123) seeded on first
run; /api routes for auth, libraries, media, stream and admin
panels. WebSocket scan-progress events at /api/ws.
- Frontend: React 18 + Vite 5 + Tailwind 3.4 + Zustand + axios +
react-router 6 + lucide-react + framer-motion + hls.js. Pages for
Login / Home / Library / Search / MediaDetail / Player / Admin
(Library, Users, Settings tabs).
- Distribution: multi-arch Dockerfile (frontend -> backend -> Alpine
runtime), docker-compose.yml, GitHub Actions for CI and GHCR
publish, Makefile, env-prefixed config (MEDIASTATION_*).
- Docs: README, CONTRIBUTING, .env.example, config.example.yaml.
Backend builds, vets and tests pass. Frontend builds via tsc -b && vite
build (250 kB JS / 16 kB CSS, gzipped 84 / 4 kB).
This commit is contained in:
@@ -0,0 +1,163 @@
|
||||
// Package service — authentication / user management.
|
||||
package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/golang-jwt/jwt/v5"
|
||||
"go.uber.org/zap"
|
||||
"golang.org/x/crypto/bcrypt"
|
||||
|
||||
"github.com/ShukeBta/MediaStationGo/internal/config"
|
||||
"github.com/ShukeBta/MediaStationGo/internal/middleware"
|
||||
"github.com/ShukeBta/MediaStationGo/internal/model"
|
||||
"github.com/ShukeBta/MediaStationGo/internal/repository"
|
||||
)
|
||||
|
||||
// AuthService handles registration, login, and JWT issuance.
|
||||
type AuthService struct {
|
||||
cfg *config.Config
|
||||
log *zap.Logger
|
||||
repo *repository.Container
|
||||
}
|
||||
|
||||
// NewAuthService is the constructor.
|
||||
func NewAuthService(cfg *config.Config, log *zap.Logger, repo *repository.Container) *AuthService {
|
||||
return &AuthService{cfg: cfg, log: log, repo: repo}
|
||||
}
|
||||
|
||||
// Common service-level errors.
|
||||
var (
|
||||
ErrInvalidCredentials = errors.New("invalid username or password")
|
||||
ErrUsernameTaken = errors.New("username already taken")
|
||||
)
|
||||
|
||||
// SeedAdmin makes sure at least one admin user exists. It mirrors the
|
||||
// MediaStation behaviour: if no admin row is found we create
|
||||
// `admin / admin123` (overridable through ADMIN_INITIAL_PASSWORD) and warn.
|
||||
func (s *AuthService) SeedAdmin(ctx context.Context) error {
|
||||
n, err := s.repo.User.CountAdmins(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if n > 0 {
|
||||
return nil
|
||||
}
|
||||
pwd := os.Getenv("ADMIN_INITIAL_PASSWORD")
|
||||
if pwd == "" {
|
||||
pwd = "admin123"
|
||||
}
|
||||
hash, err := hashPassword(pwd)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
user := &model.User{
|
||||
Username: "admin",
|
||||
PasswordHash: hash,
|
||||
Role: "admin",
|
||||
ForcePasswordReset: pwd == "admin123",
|
||||
}
|
||||
if err := s.repo.User.Create(ctx, user); err != nil {
|
||||
return err
|
||||
}
|
||||
s.log.Warn("default admin created — change the password after first login",
|
||||
zap.String("username", "admin"),
|
||||
zap.String("password_source", "ADMIN_INITIAL_PASSWORD or admin123"),
|
||||
)
|
||||
return nil
|
||||
}
|
||||
|
||||
// Register creates a new user. The first registered user is auto-promoted to
|
||||
// admin to support fresh installs that did not run SeedAdmin.
|
||||
func (s *AuthService) Register(ctx context.Context, username, password string) (*model.User, error) {
|
||||
username = strings.TrimSpace(username)
|
||||
if username == "" || password == "" {
|
||||
return nil, fmt.Errorf("username and password required")
|
||||
}
|
||||
if existing, err := s.repo.User.FindByUsername(ctx, username); err != nil {
|
||||
return nil, err
|
||||
} else if existing != nil {
|
||||
return nil, ErrUsernameTaken
|
||||
}
|
||||
hash, err := hashPassword(password)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
role := "user"
|
||||
if n, err := s.repo.User.CountAdmins(ctx); err == nil && n == 0 {
|
||||
role = "admin"
|
||||
}
|
||||
u := &model.User{Username: username, PasswordHash: hash, Role: role}
|
||||
if err := s.repo.User.Create(ctx, u); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return u, nil
|
||||
}
|
||||
|
||||
// Login validates credentials and returns the user + a fresh JWT.
|
||||
func (s *AuthService) Login(ctx context.Context, username, password string) (*model.User, string, error) {
|
||||
u, err := s.repo.User.FindByUsername(ctx, username)
|
||||
if err != nil {
|
||||
return nil, "", err
|
||||
}
|
||||
if u == nil {
|
||||
return nil, "", ErrInvalidCredentials
|
||||
}
|
||||
if err := bcrypt.CompareHashAndPassword([]byte(u.PasswordHash), []byte(password)); err != nil {
|
||||
return nil, "", ErrInvalidCredentials
|
||||
}
|
||||
token, err := s.IssueToken(u)
|
||||
if err != nil {
|
||||
return nil, "", err
|
||||
}
|
||||
_ = s.repo.User.TouchLogin(ctx, u.ID)
|
||||
return u, token, nil
|
||||
}
|
||||
|
||||
// ChangePassword updates the user password if the old one matches.
|
||||
func (s *AuthService) ChangePassword(ctx context.Context, userID, oldPwd, newPwd string) error {
|
||||
u, err := s.repo.User.FindByID(ctx, userID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if u == nil {
|
||||
return ErrInvalidCredentials
|
||||
}
|
||||
if err := bcrypt.CompareHashAndPassword([]byte(u.PasswordHash), []byte(oldPwd)); err != nil {
|
||||
return ErrInvalidCredentials
|
||||
}
|
||||
hash, err := hashPassword(newPwd)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return s.repo.User.UpdatePassword(ctx, userID, hash)
|
||||
}
|
||||
|
||||
// IssueToken signs a JWT for the given user (24h validity).
|
||||
func (s *AuthService) IssueToken(u *model.User) (string, error) {
|
||||
claims := middleware.Claims{
|
||||
UserID: u.ID,
|
||||
Role: u.Role,
|
||||
RegisteredClaims: jwt.RegisteredClaims{
|
||||
IssuedAt: jwt.NewNumericDate(time.Now()),
|
||||
ExpiresAt: jwt.NewNumericDate(time.Now().Add(24 * time.Hour)),
|
||||
Issuer: "mediastationgo",
|
||||
Subject: u.ID,
|
||||
},
|
||||
}
|
||||
t := jwt.NewWithClaims(jwt.SigningMethodHS256, claims)
|
||||
return t.SignedString([]byte(s.cfg.Secrets.JWTSecret))
|
||||
}
|
||||
|
||||
func hashPassword(p string) (string, error) {
|
||||
h, err := bcrypt.GenerateFromPassword([]byte(p), bcrypt.DefaultCost)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
return string(h), nil
|
||||
}
|
||||
@@ -0,0 +1,88 @@
|
||||
// Package service — library / media bookkeeping.
|
||||
package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
|
||||
"go.uber.org/zap"
|
||||
|
||||
"github.com/ShukeBta/MediaStationGo/internal/config"
|
||||
"github.com/ShukeBta/MediaStationGo/internal/model"
|
||||
"github.com/ShukeBta/MediaStationGo/internal/repository"
|
||||
)
|
||||
|
||||
// MediaService offers high-level CRUD over libraries and media items.
|
||||
type MediaService struct {
|
||||
cfg *config.Config
|
||||
log *zap.Logger
|
||||
repo *repository.Container
|
||||
}
|
||||
|
||||
// NewMediaService is the constructor.
|
||||
func NewMediaService(cfg *config.Config, log *zap.Logger, repo *repository.Container) *MediaService {
|
||||
return &MediaService{cfg: cfg, log: log, repo: repo}
|
||||
}
|
||||
|
||||
// CreateLibrary persists a library after validating that its path exists.
|
||||
func (s *MediaService) CreateLibrary(ctx context.Context, name, path, kind string) (*model.Library, error) {
|
||||
if name == "" || path == "" {
|
||||
return nil, errors.New("name and path required")
|
||||
}
|
||||
abs, err := filepath.Abs(path)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("invalid path: %w", err)
|
||||
}
|
||||
if info, err := os.Stat(abs); err != nil || !info.IsDir() {
|
||||
return nil, fmt.Errorf("path is not an accessible directory: %s", abs)
|
||||
}
|
||||
if kind == "" {
|
||||
kind = "movie"
|
||||
}
|
||||
lib := &model.Library{Name: name, Path: abs, Type: kind, Enabled: true}
|
||||
if err := s.repo.Library.Create(ctx, lib); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return lib, nil
|
||||
}
|
||||
|
||||
// ListLibraries returns every library configured on the server.
|
||||
func (s *MediaService) ListLibraries(ctx context.Context) ([]model.Library, error) {
|
||||
return s.repo.Library.List(ctx)
|
||||
}
|
||||
|
||||
// DeleteLibrary removes a library and its media rows. The on-disk files are
|
||||
// left untouched.
|
||||
func (s *MediaService) DeleteLibrary(ctx context.Context, id string) error {
|
||||
if err := s.repo.Media.DeleteByLibrary(ctx, id); err != nil {
|
||||
return err
|
||||
}
|
||||
return s.repo.Library.Delete(ctx, id)
|
||||
}
|
||||
|
||||
// ListMedia paginates media items inside a library.
|
||||
func (s *MediaService) ListMedia(ctx context.Context, libraryID string, page, pageSize int) ([]model.Media, int64, error) {
|
||||
if pageSize <= 0 || pageSize > 200 {
|
||||
pageSize = 50
|
||||
}
|
||||
if page < 1 {
|
||||
page = 1
|
||||
}
|
||||
return s.repo.Media.ListByLibrary(ctx, libraryID, (page-1)*pageSize, pageSize)
|
||||
}
|
||||
|
||||
// SearchMedia performs a simple LIKE search across titles.
|
||||
func (s *MediaService) SearchMedia(ctx context.Context, query string, limit int) ([]model.Media, error) {
|
||||
if limit <= 0 || limit > 200 {
|
||||
limit = 50
|
||||
}
|
||||
return s.repo.Media.Search(ctx, query, limit)
|
||||
}
|
||||
|
||||
// GetMedia returns a single media row.
|
||||
func (s *MediaService) GetMedia(ctx context.Context, id string) (*model.Media, error) {
|
||||
return s.repo.Media.FindByID(ctx, id)
|
||||
}
|
||||
@@ -0,0 +1,111 @@
|
||||
// Package service — filesystem scanner.
|
||||
//
|
||||
// ScannerService walks the configured library roots looking for video files,
|
||||
// then upserts a model.Media row per file. A future iteration will plug
|
||||
// ffprobe / a metadata-provider chain on top of this skeleton, but the
|
||||
// scaffold keeps the surface narrow and synchronous so handlers can call
|
||||
// "POST /api/libraries/:id/scan" today.
|
||||
package service
|
||||
|
||||
import (
|
||||
"context"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
|
||||
"go.uber.org/zap"
|
||||
|
||||
"github.com/ShukeBta/MediaStationGo/internal/config"
|
||||
"github.com/ShukeBta/MediaStationGo/internal/model"
|
||||
"github.com/ShukeBta/MediaStationGo/internal/repository"
|
||||
)
|
||||
|
||||
// videoExtensions lists the file extensions treated as media. Matches the
|
||||
// MediaStation Python defaults.
|
||||
var videoExtensions = map[string]struct{}{
|
||||
".mkv": {},
|
||||
".mp4": {},
|
||||
".m4v": {},
|
||||
".avi": {},
|
||||
".mov": {},
|
||||
".webm": {},
|
||||
".ts": {},
|
||||
".rmvb": {},
|
||||
".rm": {},
|
||||
".3gp": {},
|
||||
".mpg": {},
|
||||
".mpeg": {},
|
||||
".strm": {},
|
||||
}
|
||||
|
||||
// ScannerService walks libraries on disk and upserts model.Media rows.
|
||||
type ScannerService struct {
|
||||
cfg *config.Config
|
||||
log *zap.Logger
|
||||
repo *repository.Container
|
||||
hub *Hub
|
||||
}
|
||||
|
||||
// NewScannerService is the constructor.
|
||||
func NewScannerService(cfg *config.Config, log *zap.Logger, repo *repository.Container, hub *Hub) *ScannerService {
|
||||
return &ScannerService{cfg: cfg, log: log, repo: repo, hub: hub}
|
||||
}
|
||||
|
||||
// ScanResult summarises a scan run.
|
||||
type ScanResult struct {
|
||||
LibraryID string `json:"library_id"`
|
||||
Visited int `json:"visited"`
|
||||
Added int `json:"added"`
|
||||
}
|
||||
|
||||
// ScanLibrary walks the library root and persists discovered media files.
|
||||
//
|
||||
// This is a synchronous skeleton: large libraries should call it in a
|
||||
// goroutine. WebSocket progress events are pushed to the hub on the
|
||||
// "scan" topic so the React UI can display a progress indicator.
|
||||
func (s *ScannerService) ScanLibrary(ctx context.Context, libraryID string) (*ScanResult, error) {
|
||||
lib, err := s.repo.Library.FindByID(ctx, libraryID)
|
||||
if err != nil || lib == nil {
|
||||
return nil, err
|
||||
}
|
||||
res := &ScanResult{LibraryID: lib.ID}
|
||||
walkFn := func(path string, info walkInfo) error {
|
||||
if info.isDir {
|
||||
return nil
|
||||
}
|
||||
ext := strings.ToLower(filepath.Ext(path))
|
||||
if _, ok := videoExtensions[ext]; !ok {
|
||||
return nil
|
||||
}
|
||||
res.Visited++
|
||||
title := strings.TrimSuffix(filepath.Base(path), ext)
|
||||
m := &model.Media{
|
||||
LibraryID: lib.ID,
|
||||
Title: title,
|
||||
Path: path,
|
||||
SizeBytes: info.size,
|
||||
Container: strings.TrimPrefix(ext, "."),
|
||||
}
|
||||
if err := s.repo.Media.Upsert(ctx, m); err != nil {
|
||||
s.log.Warn("upsert media failed", zap.String("path", path), zap.Error(err))
|
||||
return nil
|
||||
}
|
||||
res.Added++
|
||||
s.hub.Publish("scan", map[string]any{
|
||||
"library_id": lib.ID,
|
||||
"path": path,
|
||||
"visited": res.Visited,
|
||||
"added": res.Added,
|
||||
})
|
||||
return nil
|
||||
}
|
||||
if err := walk(lib.Path, walkFn); err != nil {
|
||||
return res, err
|
||||
}
|
||||
s.hub.Publish("scan", map[string]any{
|
||||
"library_id": lib.ID,
|
||||
"finished": true,
|
||||
"visited": res.Visited,
|
||||
"added": res.Added,
|
||||
})
|
||||
return res, nil
|
||||
}
|
||||
@@ -0,0 +1,48 @@
|
||||
// Package service contains the business logic of MediaStationGo. Handlers
|
||||
// deserialize the HTTP request, call into a Service method, then serialize
|
||||
// the response. Services own all cross-cutting policy (auth, scanning,
|
||||
// transcoding, etc.) and never deal with HTTP types directly.
|
||||
package service
|
||||
|
||||
import (
|
||||
"go.uber.org/zap"
|
||||
|
||||
"github.com/ShukeBta/MediaStationGo/internal/config"
|
||||
"github.com/ShukeBta/MediaStationGo/internal/repository"
|
||||
)
|
||||
|
||||
// Container holds every service initialized at startup. Handlers receive a
|
||||
// pointer to it and pick the relevant fields.
|
||||
type Container struct {
|
||||
Cfg *config.Config
|
||||
Log *zap.Logger
|
||||
Repo *repository.Container
|
||||
WSHub *Hub
|
||||
Auth *AuthService
|
||||
Media *MediaService
|
||||
Scan *ScannerService
|
||||
Stream *StreamService
|
||||
}
|
||||
|
||||
// New builds the service container.
|
||||
func New(cfg *config.Config, log *zap.Logger, repos *repository.Container) *Container {
|
||||
hub := NewHub(log)
|
||||
go hub.Run()
|
||||
return &Container{
|
||||
Cfg: cfg,
|
||||
Log: log,
|
||||
Repo: repos,
|
||||
WSHub: hub,
|
||||
Auth: NewAuthService(cfg, log, repos),
|
||||
Media: NewMediaService(cfg, log, repos),
|
||||
Scan: NewScannerService(cfg, log, repos, hub),
|
||||
Stream: NewStreamService(cfg, log, repos),
|
||||
}
|
||||
}
|
||||
|
||||
// Close releases any resources held by services (e.g. the websocket hub).
|
||||
func (c *Container) Close() {
|
||||
if c.WSHub != nil {
|
||||
c.WSHub.Stop()
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,58 @@
|
||||
// Package service — direct-play / range request streaming.
|
||||
package service
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"net/http"
|
||||
"os"
|
||||
|
||||
"go.uber.org/zap"
|
||||
|
||||
"github.com/ShukeBta/MediaStationGo/internal/config"
|
||||
"github.com/ShukeBta/MediaStationGo/internal/repository"
|
||||
)
|
||||
|
||||
// StreamService serves media files with proper Range support so browsers can
|
||||
// seek into the stream.
|
||||
//
|
||||
// HLS / on-demand transcoding is intentionally omitted from this initial
|
||||
// scaffold. The HTTP handler returns 501 (NotImplemented) for that path,
|
||||
// while direct-play already works for browser-friendly containers (mp4 /
|
||||
// webm / m4v).
|
||||
type StreamService struct {
|
||||
cfg *config.Config
|
||||
log *zap.Logger
|
||||
repo *repository.Container
|
||||
}
|
||||
|
||||
// NewStreamService is the constructor.
|
||||
func NewStreamService(cfg *config.Config, log *zap.Logger, repo *repository.Container) *StreamService {
|
||||
return &StreamService{cfg: cfg, log: log, repo: repo}
|
||||
}
|
||||
|
||||
// ErrMediaNotFound is returned when the media row or its file is missing.
|
||||
var ErrMediaNotFound = errors.New("media not found")
|
||||
|
||||
// ServeFile streams the file backing the given media ID using
|
||||
// http.ServeContent so HEAD / Range / If-Modified-Since are handled for free.
|
||||
func (s *StreamService) ServeFile(w http.ResponseWriter, r *http.Request, mediaID string) error {
|
||||
m, err := s.repo.Media.FindByID(r.Context(), mediaID)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if m == nil {
|
||||
return ErrMediaNotFound
|
||||
}
|
||||
f, err := os.Open(m.Path)
|
||||
if err != nil {
|
||||
return ErrMediaNotFound
|
||||
}
|
||||
defer f.Close()
|
||||
stat, err := f.Stat()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
w.Header().Set("Accept-Ranges", "bytes")
|
||||
http.ServeContent(w, r, stat.Name(), stat.ModTime(), f)
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,35 @@
|
||||
// Package service — internal filesystem walker shared by scanner / watcher.
|
||||
package service
|
||||
|
||||
import (
|
||||
"io/fs"
|
||||
"path/filepath"
|
||||
)
|
||||
|
||||
// walkInfo is a tiny abstraction over os.FileInfo so that callers do not
|
||||
// need to depend on os/io packages directly.
|
||||
type walkInfo struct {
|
||||
isDir bool
|
||||
size int64
|
||||
}
|
||||
|
||||
// walk traverses root depth-first calling fn for every entry. Hidden
|
||||
// directories (starting with ".") are skipped.
|
||||
func walk(root string, fn func(string, walkInfo) error) error {
|
||||
return filepath.WalkDir(root, func(path string, d fs.DirEntry, err error) error {
|
||||
if err != nil {
|
||||
return nil // best effort — keep walking
|
||||
}
|
||||
name := d.Name()
|
||||
if d.IsDir() && name != "." && len(name) > 1 && name[0] == '.' {
|
||||
return filepath.SkipDir
|
||||
}
|
||||
info := walkInfo{isDir: d.IsDir()}
|
||||
if !d.IsDir() {
|
||||
if fi, err := d.Info(); err == nil {
|
||||
info.size = fi.Size()
|
||||
}
|
||||
}
|
||||
return fn(path, info)
|
||||
})
|
||||
}
|
||||
@@ -0,0 +1,127 @@
|
||||
// Package service — WebSocket hub used to push real-time progress events
|
||||
// (scan / scrape / transcode) to subscribed clients.
|
||||
package service
|
||||
|
||||
import (
|
||||
"sync"
|
||||
|
||||
"go.uber.org/zap"
|
||||
)
|
||||
|
||||
// Event is the JSON payload pushed to subscribers.
|
||||
type Event struct {
|
||||
Topic string `json:"topic"`
|
||||
Payload any `json:"payload"`
|
||||
}
|
||||
|
||||
// Subscriber is a single connected client; the hub writes events into Out
|
||||
// and closes Done when the connection should be torn down.
|
||||
type Subscriber struct {
|
||||
ID string
|
||||
Out chan Event
|
||||
topics map[string]struct{}
|
||||
}
|
||||
|
||||
// Hub is a fan-out broker: services publish on a topic and every subscriber
|
||||
// that opted into that topic receives the event.
|
||||
type Hub struct {
|
||||
log *zap.Logger
|
||||
mu sync.RWMutex
|
||||
subs map[string]*Subscriber
|
||||
in chan Event
|
||||
stop chan struct{}
|
||||
closed bool
|
||||
}
|
||||
|
||||
// NewHub builds a Hub. Caller must invoke Run in its own goroutine.
|
||||
func NewHub(log *zap.Logger) *Hub {
|
||||
return &Hub{
|
||||
log: log,
|
||||
subs: make(map[string]*Subscriber),
|
||||
in: make(chan Event, 256),
|
||||
stop: make(chan struct{}),
|
||||
}
|
||||
}
|
||||
|
||||
// Run is the blocking event loop. Publish events with Hub.Publish.
|
||||
func (h *Hub) Run() {
|
||||
for {
|
||||
select {
|
||||
case <-h.stop:
|
||||
return
|
||||
case ev := <-h.in:
|
||||
h.fanout(ev)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Stop terminates the hub goroutine and disconnects every subscriber.
|
||||
func (h *Hub) Stop() {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
if h.closed {
|
||||
return
|
||||
}
|
||||
h.closed = true
|
||||
close(h.stop)
|
||||
for _, sub := range h.subs {
|
||||
close(sub.Out)
|
||||
}
|
||||
h.subs = nil
|
||||
}
|
||||
|
||||
// Publish sends an event to every interested subscriber. Non-blocking: the
|
||||
// event is dropped if the hub is full to avoid stalling the producer.
|
||||
func (h *Hub) Publish(topic string, payload any) {
|
||||
select {
|
||||
case h.in <- Event{Topic: topic, Payload: payload}:
|
||||
default:
|
||||
h.log.Warn("ws hub overflow, dropping event", zap.String("topic", topic))
|
||||
}
|
||||
}
|
||||
|
||||
// Subscribe registers a new connection for a given topic set. Pass an empty
|
||||
// list to receive every topic.
|
||||
func (h *Hub) Subscribe(id string, topics []string) *Subscriber {
|
||||
sub := &Subscriber{
|
||||
ID: id,
|
||||
Out: make(chan Event, 32),
|
||||
topics: map[string]struct{}{},
|
||||
}
|
||||
for _, t := range topics {
|
||||
sub.topics[t] = struct{}{}
|
||||
}
|
||||
h.mu.Lock()
|
||||
h.subs[id] = sub
|
||||
h.mu.Unlock()
|
||||
return sub
|
||||
}
|
||||
|
||||
// Unsubscribe disconnects the subscriber and closes its outbound channel.
|
||||
func (h *Hub) Unsubscribe(id string) {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
sub, ok := h.subs[id]
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
delete(h.subs, id)
|
||||
close(sub.Out)
|
||||
}
|
||||
|
||||
func (h *Hub) fanout(ev Event) {
|
||||
h.mu.RLock()
|
||||
defer h.mu.RUnlock()
|
||||
for _, sub := range h.subs {
|
||||
if len(sub.topics) > 0 {
|
||||
if _, ok := sub.topics[ev.Topic]; !ok {
|
||||
continue
|
||||
}
|
||||
}
|
||||
select {
|
||||
case sub.Out <- ev:
|
||||
default:
|
||||
// Slow consumer: drop the event for this subscriber.
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user