LBRYCommunity/tracker
1
0
Fork 0
Code Issues 1 Pull requests Projects Releases Packages Wiki Activity

Merge pull request #21 from pushrax/storage-reorg

Browse source 
package reorganizations
This commit is contained in:
Jimmy Zelinskie 2013-10-21 15:08:12 -07:00
commit 54e769a8f2
21 changed files with 388 additions and 386 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

85
cache/cache.go vendored
View file

@ -1,85 +0,0 @@
// Copyright 2013 The Chihaya Authors. All rights reserved.
// Use of this source code is governed by the BSD 2-Clause license,
// which can be found in the LICENSE file.
// Package cache provides a generic interface for manipulating a
// BitTorrent tracker's fast moving data.
package cache
import (
"fmt"
"github.com/pushrax/chihaya/config"
"github.com/pushrax/chihaya/models"
)
var (
drivers = make(map[string]Driver)
)
type Driver interface {
New(*config.DataStore) Pool
}
// Register makes a database driver available by the provided name.
// If Register is called twice with the same name or if driver is nil,
// it panics.
func Register(name string, driver Driver) {
if driver == nil {
panic("cache: Register driver is nil")
}
if _, dup := drivers[name]; dup {
panic("cache: Register called twice for driver " + name)
}
drivers[name] = driver
}
// Open creates a pool of data store connections specified by a storage configuration.
func Open(conf *config.DataStore) (Pool, error) {
driver, ok := drivers[conf.Driver]
if !ok {
return nil, fmt.Errorf(
"cache: unknown driver %q (forgotten import?)",
conf.Driver,
)
}
pool := driver.New(conf)
return pool, nil
}
// Pool represents a thread-safe pool of connections to the data store
// that can be used to obtain transactions.
type Pool interface {
Close() error
Get() (Tx, error)
}
// The transmit object is the interface to add, remove and modify
// data in the cache
type Tx interface {
// Reads
FindUser(passkey string) (*models.User, bool, error)
FindTorrent(infohash string) (*models.Torrent, bool, error)
ClientWhitelisted(peerID string) (bool, error)
// Writes
RecordSnatch(u *models.User, t *models.Torrent) error
MarkActive(t *models.Torrent) error
AddLeecher(t *models.Torrent, p *models.Peer) error
AddSeeder(t *models.Torrent, p *models.Peer) error
RemoveLeecher(t *models.Torrent, p *models.Peer) error
RemoveSeeder(t *models.Torrent, p *models.Peer) error
SetLeecher(t *models.Torrent, p *models.Peer) error
SetSeeder(t *models.Torrent, p *models.Peer) error
IncrementSlots(u *models.User) error
DecrementSlots(u *models.User) error
LeecherFinished(t *models.Torrent, p *models.Peer) error
// Priming / Testing
AddTorrent(t *models.Torrent) error
RemoveTorrent(t *models.Torrent) error
AddUser(u *models.User) error
RemoveUser(u *models.User) error
WhitelistClient(peerID string) error
UnWhitelistClient(peerID string) error
}

6
main.go
View file

@ -15,9 +15,9 @@ import (
"github.com/pushrax/chihaya/config"
"github.com/pushrax/chihaya/server"
_ "github.com/pushrax/chihaya/cache/redis"
_ "github.com/pushrax/chihaya/storage/batter"
_ "github.com/pushrax/chihaya/storage/gazelle"
_ "github.com/pushrax/chihaya/storage/backend/batter"
_ "github.com/pushrax/chihaya/storage/backend/gazelle"
_ "github.com/pushrax/chihaya/storage/tracker/redis"
)
var (

52
models/models.go
View file

@ -1,52 +0,0 @@
// Copyright 2013 The Chihaya Authors. All rights reserved.
// Use of this source code is governed by the BSD 2-Clause license,
// which can be found in the LICENSE file.
package models
import (
"strconv"
)
type Peer struct {
ID string `json:"id"`
UserID uint64 `json:"user_id"`
TorrentID uint64 `json:"torrent_id"`
IP string `json:"ip"`
Port uint64 `json:"port"`
Uploaded uint64 `json:"uploaded"`
Downloaded uint64 `json:"downloaded`
Left uint64 `json:"left"`
LastAnnounce int64 `json:"last_announce"`
}
func PeerMapKey(peer *Peer) string {
return peer.ID + ":" + strconv.FormatUint(peer.UserID, 36)
}
type Torrent struct {
ID uint64 `json:"id"`
Infohash string `json:"infohash"`
Active bool `json:"active"`
Seeders map[string]Peer `json:"seeders"`
Leechers map[string]Peer `json:"leechers"`
Snatches uint64 `json:"snatches"`
UpMultiplier float64 `json:"up_multiplier"`
DownMultiplier float64 `json:"down_multiplier"`
LastAction int64 `json:"last_action"`
}
type User struct {
ID uint64 `json:"id"`
Passkey string `json:"passkey"`
UpMultiplier float64 `json:"up_multiplier"`
DownMultiplier float64 `json:"down_multiplier"`
Slots int64 `json:"slots"`
SlotsUsed int64 `json:"slots_used"`
Snatches uint64 `json:"snatches"`
}

12
server/announce.go
View file

@ -12,7 +12,7 @@ import (
"strconv"
"time"
"github.com/pushrax/chihaya/models"
"github.com/pushrax/chihaya/storage"
)
func (s Server) serveAnnounce(w http.ResponseWriter, r *http.Request) {
@ -69,7 +69,7 @@ func (s Server) serveAnnounce(w http.ResponseWriter, r *http.Request) {
}
// Create a new peer object from the request
peer := &models.Peer{
peer := &storage.Peer{
ID: peerID,
UserID: user.ID,
TorrentID: torrent.ID,
@ -82,8 +82,8 @@ func (s Server) serveAnnounce(w http.ResponseWriter, r *http.Request) {
}
// Look for the user in in the pool of seeders and leechers
_, seeder := torrent.Seeders[models.PeerMapKey(peer)]
_, leecher := torrent.Leechers[models.PeerMapKey(peer)]
_, seeder := torrent.Seeders[storage.PeerMapKey(peer)]
_, leecher := torrent.Leechers[storage.PeerMapKey(peer)]
switch {
// Guarantee that no user is in both pools
@ -318,7 +318,7 @@ func minInt(a, b int) int {
return b
}
func writeSeeders(w http.ResponseWriter, t *models.Torrent, count, numWant int, compact bool) {
func writeSeeders(w http.ResponseWriter, t *storage.Torrent, count, numWant int, compact bool) {
for _, seed := range t.Seeders {
if count >= numWant {
break
@ -339,7 +339,7 @@ func writeSeeders(w http.ResponseWriter, t *models.Torrent, count, numWant int,
}
}
func writeLeechers(w http.ResponseWriter, t *models.Torrent, count, numWant int, compact bool) {
func writeLeechers(w http.ResponseWriter, t *storage.Torrent, count, numWant int, compact bool) {
for _, leech := range t.Leechers {
if count >= numWant {
break

4
server/scrape.go
View file

@ -11,7 +11,7 @@ import (
"net/http"
"path"
"github.com/pushrax/chihaya/models"
"github.com/pushrax/chihaya/storage"
)
func (s *Server) serveScrape(w http.ResponseWriter, r *http.Request) {
@ -68,7 +68,7 @@ func (s *Server) serveScrape(w http.ResponseWriter, r *http.Request) {
w.(http.Flusher).Flush()
}
func writeScrapeInfo(w io.Writer, torrent *models.Torrent) {
func writeScrapeInfo(w io.Writer, torrent *storage.Torrent) {
io.WriteString(w, "d")
writeBencoded(w, "complete")
writeBencoded(w, len(torrent.Seeders))

10
server/server.go
View file

@ -17,15 +17,15 @@ import (
"sync/atomic"
"time"
"github.com/pushrax/chihaya/cache"
"github.com/pushrax/chihaya/config"
"github.com/pushrax/chihaya/models"
"github.com/pushrax/chihaya/storage"
"github.com/pushrax/chihaya/storage/tracker"
)
type Server struct {
conf *config.Config
listener net.Listener
dbConnPool cache.Pool
dbConnPool tracker.Pool
serving bool
startTime time.Time
@ -41,7 +41,7 @@ type Server struct {
}
func New(conf *config.Config) (*Server, error) {
pool, err := cache.Open(&conf.Cache)
pool, err := tracker.Open(&conf.Cache)
if err != nil {
return nil, err
}
@ -133,7 +133,7 @@ func fail(err error, w http.ResponseWriter, r *http.Request) {
w.(http.Flusher).Flush()
}
func validateUser(tx cache.Tx, dir string) (*models.User, error) {
func validateUser(tx tracker.Conn, dir string) (*storage.User, error) {
if len(dir) != 34 {
return nil, errors.New("Passkey is invalid")
}

4
server/stats_test.go
View file

@ -13,8 +13,8 @@ import (
"github.com/pushrax/chihaya/config"
_ "github.com/pushrax/chihaya/cache/redis"
_ "github.com/pushrax/chihaya/storage/batter"
_ "github.com/pushrax/chihaya/storage/backend/batter"
_ "github.com/pushrax/chihaya/storage/tracker/redis"
)
func newTestServer() (*Server, error) {

97
storage/backend/backend.go Normal file
View file

@ -0,0 +1,97 @@
// Copyright 2013 The Chihaya Authors. All rights reserved.
// Use of this source code is governed by the BSD 2-Clause license,
// which can be found in the LICENSE file.
// Package backend provides a generic interface for manipulating a
// BitTorrent tracker's backend data (usually for a web application).
package backend
import (
"fmt"
"github.com/pushrax/chihaya/config"
"github.com/pushrax/chihaya/storage"
)
var drivers = make(map[string]Driver)
type Driver interface {
New(*config.DataStore) Conn
}
// Register makes a database driver available by the provided name.
// If Register is called twice with the same name or if driver is nil,
// it panics.
func Register(name string, driver Driver) {
if driver == nil {
panic("web: Register driver is nil")
}
if _, dup := drivers[name]; dup {
panic("web: Register called twice for driver " + name)
}
drivers[name] = driver
}
// Open creates a connection specified by a storage configuration.
func Open(conf *config.DataStore) (Conn, error) {
driver, ok := drivers[conf.Driver]
if !ok {
return nil, fmt.Errorf(
"web: unknown driver %q (forgotten import?)",
conf.Driver,
)
}
pool := driver.New(conf)
return pool, nil
}
// Conn represents a connection to the data store.
type Conn interface {
// Start is called once when the server starts.
// It starts any necessary goroutines a given driver requires, and sets
// up the driver's initial state
Start() error
// Close terminates connections to the database(s) and gracefully shuts
// down the driver
Close() error
// RecordAnnounce is called once per announce, and is passed the delta in
// statistics for the client peer since its last announce.
RecordAnnounce(delta *AnnounceDelta) error
// LoadTorrents fetches and returns the specified torrents.
LoadTorrents(ids []uint64) ([]*storage.Torrent, error)
// LoadAllTorrents fetches and returns all torrents.
LoadAllTorrents() ([]*storage.Torrent, error)
// LoadUsers fetches and returns the specified users.
LoadUsers(ids []uint64) ([]*storage.User, error)
// LoadAllUsers fetches and returns all users.
LoadAllUsers(ids []uint64) ([]*storage.User, error)
}
// AnnounceDelta contains a difference in statistics for a peer.
// It is used for communicating changes to be recorded by the driver.
type AnnounceDelta struct {
Peer *storage.Peer
Torrent *storage.Torrent
User *storage.User
// Created is true if this announce created a new peer or changed an existing peer's address
Created bool
// Uploaded contains the raw upload delta for this announce, in bytes
Uploaded uint64
// Downloaded contains the raw download delta for this announce, in bytes
Downloaded uint64
// Timestamp is the unix timestamp this announce occurred at
Timestamp float64
// Snatched is true if this announce completed the download
Snatched bool
}

8
storage/batter/batter.go → storage/backend/batter/batter.go
View file

@ -11,14 +11,14 @@ import (
"fmt"
"github.com/pushrax/chihaya/config"
"github.com/pushrax/chihaya/storage"
"github.com/pushrax/chihaya/storage/backend"
_ "github.com/bmizerany/pq"
)
type driver struct{}
func (d *driver) New(conf *config.DataStore) storage.Conn {
func (d *driver) New(conf *config.DataStore) backend.Conn {
dsn := fmt.Sprintf(
"host=%s user=%s password=%s dbname=%s",
conf.Host,
@ -47,10 +47,10 @@ func (c *Conn) Start() error {
return nil
}
func (c *Conn) RecordAnnounce(delta *storage.AnnounceDelta) error {
func (c *Conn) RecordAnnounce(delta *backend.AnnounceDelta) error {
return nil
}
func init() {
storage.Register("batter", &driver{})
backend.Register("batter", &driver{})
}

25
storage/backend/batter/load.go Normal file
View file

@ -0,0 +1,25 @@
// Copyright 2013 The Chihaya Authors. All rights reserved.
// Use of this source code is governed by the BSD 2-Clause license,
// which can be found in the LICENSE file.
package batter
import (
"github.com/pushrax/chihaya/storage"
)
func (c *Conn) LoadTorrents(ids []uint64) ([]*storage.Torrent, error) {
return nil, nil
}
func (c *Conn) LoadAllTorrents() ([]*storage.Torrent, error) {
return nil, nil
}
func (c *Conn) LoadUsers(ids []uint64) ([]*storage.User, error) {
return nil, nil
}
func (c *Conn) LoadAllUsers(ids []uint64) ([]*storage.User, error) {
return nil, nil
}

0
storage/gazelle/flush.go → storage/backend/gazelle/flush.go
View file

8
storage/gazelle/gazelle.go → storage/backend/gazelle/gazelle.go
View file

@ -12,14 +12,14 @@ import (
"sync"
"github.com/pushrax/chihaya/config"
"github.com/pushrax/chihaya/storage"
"github.com/pushrax/chihaya/storage/backend"
_ "github.com/go-sql-driver/mysql"
)
type driver struct{}
func (d *driver) New(conf *config.DataStore) storage.Conn {
func (d *driver) New(conf *config.DataStore) backend.Conn {
dsn := fmt.Sprintf(
"%s:%s@%s:%s/%s?charset=utf8mb4,utf8",
conf.Username,
@ -77,7 +77,7 @@ func (c *Conn) Close() error {
return c.DB.Close()
}
func (c *Conn) RecordAnnounce(delta *storage.AnnounceDelta) error {
func (c *Conn) RecordAnnounce(delta *backend.AnnounceDelta) error {
snatchCount := 0
if delta.Snatched {
snatchCount = 1
@ -95,5 +95,5 @@ func (c *Conn) RecordAnnounce(delta *storage.AnnounceDelta) error {
}
func init() {
storage.Register("gazelle", &driver{})
backend.Register("gazelle", &driver{})
}

25
storage/backend/gazelle/load.go Normal file
View file

@ -0,0 +1,25 @@
// Copyright 2013 The Chihaya Authors. All rights reserved.
// Use of this source code is governed by the BSD 2-Clause license,
// which can be found in the LICENSE file.
package gazelle
import (
"github.com/pushrax/chihaya/storage"
)
func (c *Conn) LoadTorrents(ids []uint64) ([]*storage.Torrent, error) {
return nil, nil
}
func (c *Conn) LoadAllTorrents() ([]*storage.Torrent, error) {
return nil, nil
}
func (c *Conn) LoadUsers(ids []uint64) ([]*storage.User, error) {
return nil, nil
}
func (c *Conn) LoadAllUsers(ids []uint64) ([]*storage.User, error) {
return nil, nil
}

25
storage/batter/load.go
View file

@ -1,25 +0,0 @@
// Copyright 2013 The Chihaya Authors. All rights reserved.
// Use of this source code is governed by the BSD 2-Clause license,
// which can be found in the LICENSE file.
package batter
import (
"github.com/pushrax/chihaya/models"
)
func (c *Conn) LoadTorrents(ids []uint64) ([]*models.Torrent, error) {
return nil, nil
}
func (c *Conn) LoadAllTorrents() ([]*models.Torrent, error) {
return nil, nil
}
func (c *Conn) LoadUsers(ids []uint64) ([]*models.User, error) {
return nil, nil
}
func (c *Conn) LoadAllUsers(ids []uint64) ([]*models.User, error) {
return nil, nil
}

25
storage/gazelle/load.go
View file

@ -1,25 +0,0 @@
// Copyright 2013 The Chihaya Authors. All rights reserved.
// Use of this source code is governed by the BSD 2-Clause license,
// which can be found in the LICENSE file.
package gazelle
import (
"github.com/pushrax/chihaya/models"
)
func (c *Conn) LoadTorrents(ids []uint64) ([]*models.Torrent, error) {
return nil, nil
}
func (c *Conn) LoadAllTorrents() ([]*models.Torrent, error) {
return nil, nil
}
func (c *Conn) LoadUsers(ids []uint64) ([]*models.User, error) {
return nil, nil
}
func (c *Conn) LoadAllUsers(ids []uint64) ([]*models.User, error) {
return nil, nil
}

115
storage/storage.go
View file

@ -2,96 +2,53 @@
// Use of this source code is governed by the BSD 2-Clause license,
// which can be found in the LICENSE file.
// Package storage provides a generic interface for manipulating a
// BitTorrent tracker's web application data.
// Package storage implements a high-level abstraction over the multiple
// data stores used by a BitTorrent tracker.
package storage
import (
"fmt"
"github.com/pushrax/chihaya/config"
"github.com/pushrax/chihaya/models"
"strconv"
)
var drivers = make(map[string]Driver)
type Peer struct {
ID string `json:"id"`
UserID uint64 `json:"user_id"`
TorrentID uint64 `json:"torrent_id"`
type Driver interface {
New(*config.DataStore) Conn
IP string `json:"ip"`
Port uint64 `json:"port"`
Uploaded uint64 `json:"uploaded"`
Downloaded uint64 `json:"downloaded`
Left uint64 `json:"left"`
LastAnnounce int64 `json:"last_announce"`
}
// Register makes a database driver available by the provided name.
// If Register is called twice with the same name or if driver is nil,
// it panics.
func Register(name string, driver Driver) {
if driver == nil {
panic("storage: Register driver is nil")
}
if _, dup := drivers[name]; dup {
panic("storage: Register called twice for driver " + name)
}
drivers[name] = driver
func PeerMapKey(peer *Peer) string {
return peer.ID + ":" + strconv.FormatUint(peer.UserID, 36)
}
// Open creates a connection specified by a storage configuration.
func Open(conf *config.DataStore) (Conn, error) {
driver, ok := drivers[conf.Driver]
if !ok {
return nil, fmt.Errorf(
"storage: unknown driver %q (forgotten import?)",
conf.Driver,
)
}
pool := driver.New(conf)
return pool, nil
type Torrent struct {
ID uint64 `json:"id"`
Infohash string `json:"infohash"`
Active bool `json:"active"`
Seeders map[string]Peer `json:"seeders"`
Leechers map[string]Peer `json:"leechers"`
Snatches uint64 `json:"snatches"`
UpMultiplier float64 `json:"up_multiplier"`
DownMultiplier float64 `json:"down_multiplier"`
LastAction int64 `json:"last_action"`
}
// Conn represents a connection to the data store.
type Conn interface {
// Start is called once when the server starts.
// It starts any necessary goroutines a given driver requires, and sets
// up the driver's initial state
Start() error
type User struct {
ID uint64 `json:"id"`
Passkey string `json:"passkey"`
// Close terminates connections to the database(s) and gracefully shuts
// down the driver
Close() error
// RecordAnnounce is called once per announce, and is passed the delta in
// statistics for the client peer since its last announce.
RecordAnnounce(delta *AnnounceDelta) error
// LoadTorrents fetches and returns the specified torrents.
LoadTorrents(ids []uint64) ([]*models.Torrent, error)
// LoadAllTorrents fetches and returns all torrents.
LoadAllTorrents() ([]*models.Torrent, error)
// LoadUsers fetches and returns the specified users.
LoadUsers(ids []uint64) ([]*models.User, error)
// LoadAllUsers fetches and returns all users.
LoadAllUsers(ids []uint64) ([]*models.User, error)
}
// AnnounceDelta contains a difference in statistics for a peer.
// It is used for communicating changes to be recorded by the storage driver.
type AnnounceDelta struct {
Peer *models.Peer
Torrent *models.Torrent
User *models.User
// Created is true if this announce created a new peer or changed an existing peer's address
Created bool
// Uploaded contains the raw upload delta for this announce, in bytes
Uploaded uint64
// Downloaded contains the raw download delta for this announce, in bytes
Downloaded uint64
// Timestamp is the unix timestamp this announce occurred at
Timestamp float64
// Snatched is true if this announce completed the download
Snatched bool
UpMultiplier float64 `json:"up_multiplier"`
DownMultiplier float64 `json:"down_multiplier"`
Slots int64 `json:"slots"`
SlotsUsed int64 `json:"slots_used"`
Snatches uint64 `json:"snatches"`
}

114
cache/redis/redis.go → storage/tracker/redis/redis.go
View file

@ -6,7 +6,7 @@
//
// This interface is configured by a config.DataStore.
// To get a handle to this interface, call New on the initialized driver and
// then Get() on returned the cache.Pool.
// then Get() on returned the tracker.Pool.
//
// Torrents, Users, and Peers are all stored in Redis hash types. All Redis
// keys can have an optional prefix specified during configuration.
@ -30,9 +30,9 @@ import (
"github.com/garyburd/redigo/redis"
"github.com/pushrax/chihaya/cache"
"github.com/pushrax/chihaya/config"
"github.com/pushrax/chihaya/models"
"github.com/pushrax/chihaya/storage"
"github.com/pushrax/chihaya/storage/tracker"
)
var (
@ -50,8 +50,8 @@ var (
type driver struct{}
// New creates and returns a cache.Pool.
func (d *driver) New(conf *config.DataStore) cache.Pool {
// New creates and returns a tracker.Pool.
func (d *driver) New(conf *config.DataStore) tracker.Pool {
return &Pool{
conf: conf,
pool: redis.Pool{
@ -89,7 +89,7 @@ func (p *Pool) Close() error {
return p.pool.Close()
}
func (p *Pool) Get() (cache.Tx, error) {
func (p *Pool) Get() (tracker.Conn, error) {
retTx := &Tx{
conf: p.conf,
done: false,
@ -113,17 +113,17 @@ func (tx *Tx) close() {
}
// createUser takes a string slice of length 14 and returns a pointer to a new
// models.User or an error.
// storage.User or an error.
// This function is used to create a user from a Redis hash response(HGETALL).
// The order of strings the in the slice must follow the pattern:
// [<field name>, <field value>, <field name>, <field value>, ...]
// If the field value string cannot be converted to the correct type,
// createUser will return a nil user and the conversion error.
func createUser(userVals []string) (*models.User, error) {
func createUser(userVals []string) (*storage.User, error) {
if len(userVals) != 14 {
return nil, ErrCreateUser
}
var user models.User
var user storage.User
var err error
for index, userString := range userVals {
switch userString {
@ -149,7 +149,7 @@ func createUser(userVals []string) (*models.User, error) {
return &user, nil
}
// createTorrent takes a string slice of length 14 and returns a pointer to a new models.Torrent
// createTorrent takes a string slice of length 14 and returns a pointer to a new storage.Torrent
// or an error.
// This function can be used to create a torrent from a Redis hash response(HGETALL).
// The order of strings the in the slice must follow the pattern:
@ -158,11 +158,11 @@ func createUser(userVals []string) (*models.User, error) {
// If the field values cannot be converted to the correct type,
// createTorrent will return a nil user and the conversion error.
// After converting the torrent fields, the seeders and leechers are populated by redis.getPeers
func (tx *Tx) createTorrent(torrentVals []string) (*models.Torrent, error) {
func (tx *Tx) createTorrent(torrentVals []string) (*storage.Torrent, error) {
if len(torrentVals) != 14 {
return nil, ErrCreateTorrent
}
var torrent models.Torrent
var torrent storage.Torrent
var err error
for index, torrentString := range torrentVals {
switch torrentString {
@ -197,8 +197,8 @@ func (tx *Tx) createTorrent(torrentVals []string) (*models.Torrent, error) {
}
// setPeer writes or overwrites peer information, stored as a Redis hash.
// The hash fields names are the same as the JSON tags on the models.Peer struct.
func (tx *Tx) setPeer(peer *models.Peer) error {
// The hash fields names are the same as the JSON tags on the storage.Peer struct.
func (tx *Tx) setPeer(peer *storage.Peer) error {
hashKey := tx.conf.Prefix + getPeerHashKey(peer)
_, err := tx.Do("HMSET", hashKey,
"id", peer.ID,
@ -218,7 +218,7 @@ func (tx *Tx) setPeer(peer *models.Peer) error {
// and removes the peer information.
// This function calls multiple redis commands, it's not internally atomic.
// This function will not return an error if the peer to remove doesn't exist.
func (tx *Tx) removePeer(peer *models.Peer, peerTypePrefix string) error {
func (tx *Tx) removePeer(peer *storage.Peer, peerTypePrefix string) error {
setKey := tx.conf.Prefix + getPeerSetKey(peerTypePrefix, peer)
_, err := tx.Do("SREM", setKey, getPeerHashKey(peer))
if err != nil {
@ -234,14 +234,14 @@ func (tx *Tx) removePeer(peer *models.Peer, peerTypePrefix string) error {
// This function will not return an error if the peer to remove doesn't exist.
// This function will only delete the peer set if all the individual peer deletions were successful
// This function calls multiple redis commands, it's not internally atomic.
func (tx *Tx) removePeers(torrentID uint64, peers map[string]models.Peer, peerTypePrefix string) error {
func (tx *Tx) removePeers(torrentID uint64, peers map[string]storage.Peer, peerTypePrefix string) error {
for _, peer := range peers {
hashKey := tx.conf.Prefix + getPeerHashKey(&peer)
_, err := tx.Do("DEL", hashKey)
if err != nil {
return err
}
delete(peers, models.PeerMapKey(&peer))
delete(peers, storage.PeerMapKey(&peer))
}
setKey := tx.conf.Prefix + peerTypePrefix + strconv.FormatUint(torrentID, 36)
_, err := tx.Do("DEL", setKey)
@ -255,20 +255,20 @@ func (tx *Tx) removePeers(torrentID uint64, peers map[string]models.Peer, peerTy
// concatenated and delimited by colons
// This key corresponds to a Redis hash type with fields containing a peer's data.
// The peer hashkey relies on the combination of peerID, userID, and torrentID being unique.
func getPeerHashKey(peer *models.Peer) string {
func getPeerHashKey(peer *storage.Peer) string {
return peer.ID + ":" + strconv.FormatUint(peer.UserID, 36) + ":" + strconv.FormatUint(peer.TorrentID, 36)
}
// getPeerSetKey returns a string that is the peer's encoded torrentID appended to the typePrefix
// This key corresponds to a torrent's pool of leechers or seeders
func getPeerSetKey(typePrefix string, peer *models.Peer) string {
func getPeerSetKey(typePrefix string, peer *storage.Peer) string {
return typePrefix + strconv.FormatUint(peer.TorrentID, 36)
}
// addPeers adds each peer's key to the specified peer set and saves the peer's information.
// This function will not return an error if the peer already exists in the set.
// This function calls multiple redis commands, it's not internally atomic.
func (tx *Tx) addPeers(peers map[string]models.Peer, peerTypePrefix string) error {
func (tx *Tx) addPeers(peers map[string]storage.Peer, peerTypePrefix string) error {
for _, peer := range peers {
setKey := tx.conf.Prefix + getPeerSetKey(peerTypePrefix, &peer)
_, err := tx.Do("SADD", setKey, getPeerHashKey(&peer))
@ -280,17 +280,17 @@ func (tx *Tx) addPeers(peers map[string]models.Peer, peerTypePrefix string) erro
return nil
}
// createPeer takes a slice of length 9 and returns a pointer to a new models.Peer or an error.
// createPeer takes a slice of length 9 and returns a pointer to a new storage.Peer or an error.
// This function is used to create a peer from a Redis hash response(HGETALL).
// The order of strings the in the slice must follow the pattern:
// [<field name>, <field value>, <field name>, <field value>, ...]
// If the field value string cannot be converted to the correct type,
// the function will return a nil peer and the conversion error.
func createPeer(peerVals []string) (*models.Peer, error) {
func createPeer(peerVals []string) (*storage.Peer, error) {
if len(peerVals) != 18 {
return nil, ErrCreatePeer
}
var peer models.Peer
var peer storage.Peer
var err error
for index, peerString := range peerVals {
switch peerString {
@ -322,8 +322,8 @@ func createPeer(peerVals []string) (*models.Peer, error) {
// getPeers returns a map of peers from a specified torrent's peer set(seeders or leechers).
// This is a multiple action command, it's not internally atomic.
func (tx *Tx) getPeers(torrentID uint64, peerTypePrefix string) (peers map[string]models.Peer, err error) {
peers = make(map[string]models.Peer)
func (tx *Tx) getPeers(torrentID uint64, peerTypePrefix string) (peers map[string]storage.Peer, err error) {
peers = make(map[string]storage.Peer)
setKey := tx.conf.Prefix + peerTypePrefix + strconv.FormatUint(torrentID, 36)
peerStrings, err := redis.Strings(tx.Do("SMEMBERS", setKey))
if err != nil {
@ -343,15 +343,15 @@ func (tx *Tx) getPeers(torrentID uint64, peerTypePrefix string) (peers map[strin
if err != nil {
return nil, err
}
peers[models.PeerMapKey(peer)] = *peer
peers[storage.PeerMapKey(peer)] = *peer
}
return
}
// AddTorrent writes/overwrites torrent information and saves peers from both peer sets.
// The hash fields names are the same as the JSON tags on the models.Torrent struct.
// The hash fields names are the same as the JSON tags on the storage.Torrent struct.
// This is a multiple action command, it's not internally atomic.
func (tx *Tx) AddTorrent(t *models.Torrent) error {
func (tx *Tx) AddTorrent(t *storage.Torrent) error {
hashkey := tx.conf.Prefix + TorrentPrefix + t.Infohash
_, err := tx.Do("HMSET", hashkey,
"id", t.ID,
@ -379,7 +379,7 @@ func (tx *Tx) AddTorrent(t *models.Torrent) error {
// RemoveTorrent deletes the torrent's Redis hash and then deletes all peers.
// This function will not return an error if the torrent has already been removed.
// This is a multiple action command, it's not internally atomic.
func (tx *Tx) RemoveTorrent(t *models.Torrent) error {
func (tx *Tx) RemoveTorrent(t *storage.Torrent) error {
hashkey := tx.conf.Prefix + TorrentPrefix + t.Infohash
_, err := tx.Do("DEL", hashkey)
if err != nil {
@ -398,8 +398,8 @@ func (tx *Tx) RemoveTorrent(t *models.Torrent) error {
}
// AddUser writes/overwrites user information to a Redis hash.
// The hash fields names are the same as the JSON tags on the models.user struct.
func (tx *Tx) AddUser(u *models.User) error {
// The hash fields names are the same as the JSON tags on the storage.user struct.
func (tx *Tx) AddUser(u *storage.User) error {
hashkey := tx.conf.Prefix + UserPrefix + u.Passkey
_, err := tx.Do("HMSET", hashkey,
"id", u.ID,
@ -417,7 +417,7 @@ func (tx *Tx) AddUser(u *models.User) error {
// RemoveUser removes the user's hash from Redis.
// This function does not return an error if the user doesn't exist.
func (tx *Tx) RemoveUser(u *models.User) error {
func (tx *Tx) RemoveUser(u *storage.User) error {
hashkey := tx.conf.Prefix + UserPrefix + u.Passkey
_, err := tx.Do("DEL", hashkey)
if err != nil {
@ -429,7 +429,7 @@ func (tx *Tx) RemoveUser(u *models.User) error {
// FindUser returns a pointer to a new user struct and true if the user exists,
// or nil and false if the user doesn't exist.
// This function does not return an error if the torrent doesn't exist.
func (tx *Tx) FindUser(passkey string) (*models.User, bool, error) {
func (tx *Tx) FindUser(passkey string) (*storage.User, bool, error) {
hashkey := tx.conf.Prefix + UserPrefix + passkey
// Consider using HGETALL instead of HVALS here for robustness
userStrings, err := redis.Strings(tx.Do("HGETALL", hashkey))
@ -448,7 +448,7 @@ func (tx *Tx) FindUser(passkey string) (*models.User, bool, error) {
// FindTorrent returns a pointer to a new torrent struct and true if the torrent exists,
// or nil and false if the torrent doesn't exist.
// This is a multiple action command, it's not internally atomic.
func (tx *Tx) FindTorrent(infohash string) (*models.Torrent, bool, error) {
func (tx *Tx) FindTorrent(infohash string) (*storage.Torrent, bool, error) {
hashkey := tx.conf.Prefix + TorrentPrefix + infohash
torrentStrings, err := redis.Strings(tx.Do("HGETALL", hashkey))
if err != nil {
@ -491,7 +491,7 @@ func (tx *Tx) UnWhitelistClient(peerID string) error {
// RecordSnatch increments the snatch counter on the torrent and user by one.
// This modifies the arguments as well as the hash field in Redis.
// This is a multiple action command, it's not internally atomic.
func (tx *Tx) RecordSnatch(user *models.User, torrent *models.Torrent) error {
func (tx *Tx) RecordSnatch(user *storage.User, torrent *storage.Torrent) error {
torrentKey := tx.conf.Prefix + TorrentPrefix + torrent.Infohash
snatchCount, err := redis.Int(tx.Do("HINCRBY", torrentKey, "snatches", 1))
@ -512,7 +512,7 @@ func (tx *Tx) RecordSnatch(user *models.User, torrent *models.Torrent) error {
// MarkActive sets the active field of the torrent to true.
// This modifies the argument as well as the hash field in Redis.
// This function will return ErrMarkActive if the torrent does not exist.
func (tx *Tx) MarkActive(torrent *models.Torrent) error {
func (tx *Tx) MarkActive(torrent *storage.Torrent) error {
hashkey := tx.conf.Prefix + TorrentPrefix + torrent.Infohash
activeExists, err := redis.Int(tx.Do("HSET", hashkey, "active", true))
if err != nil {
@ -529,7 +529,7 @@ func (tx *Tx) MarkActive(torrent *models.Torrent) error {
// MarkInactive sets the active field of the torrent to false.
// This modifies the argument as well as the hash field in Redis.
// This function will return ErrMarkActive if the torrent does not exist.
func (tx *Tx) MarkInactive(torrent *models.Torrent) error {
func (tx *Tx) MarkInactive(torrent *storage.Torrent) error {
hashkey := tx.conf.Prefix + TorrentPrefix + torrent.Infohash
activeExists, err := redis.Int(tx.Do("HSET", hashkey, "active", false))
if err != nil {
@ -552,7 +552,7 @@ func (tx *Tx) MarkInactive(torrent *models.Torrent) error {
// This modifies the torrent argument, as well as the torrent's set and peer's hash in Redis.
// This function does not return an error if the leecher already exists.
// This is a multiple action command, it's not internally atomic.
func (tx *Tx) AddLeecher(torrent *models.Torrent, peer *models.Peer) error {
func (tx *Tx) AddLeecher(torrent *storage.Torrent, peer *storage.Peer) error {
setKey := tx.conf.Prefix + LeechersPrefix + strconv.FormatUint(torrent.ID, 36)
_, err := tx.Do("SADD", setKey, getPeerHashKey(peer))
if err != nil {
@ -563,9 +563,9 @@ func (tx *Tx) AddLeecher(torrent *models.Torrent, peer *models.Peer) error {
return err
}
if torrent.Leechers == nil {
torrent.Leechers = make(map[string]models.Peer)
torrent.Leechers = make(map[string]storage.Peer)
}
torrent.Leechers[models.PeerMapKey(peer)] = *peer
torrent.Leechers[storage.PeerMapKey(peer)] = *peer
return nil
}
@ -573,31 +573,31 @@ func (tx *Tx) AddLeecher(torrent *models.Torrent, peer *models.Peer) error {
// This modifies the torrent argument, as well as the peer's hash in Redis.
// Setting assumes that the peer is already a leecher, and only needs to be updated.
// This function does not return an error if the leecher does not exist or is not in the torrent's leecher set.
func (tx *Tx) SetLeecher(t *models.Torrent, p *models.Peer) error {
func (tx *Tx) SetLeecher(t *storage.Torrent, p *storage.Peer) error {
err := tx.setPeer(p)
if err != nil {
return err
}
t.Leechers[models.PeerMapKey(p)] = *p
t.Leechers[storage.PeerMapKey(p)] = *p
return nil
}
// RemoveLeecher removes the given peer from a torrent's leecher set.
// This modifies the torrent argument, as well as the torrent's set and peer's hash in Redis.
// This function does not return an error if the peer doesn't exist, or is not in the set.
func (tx *Tx) RemoveLeecher(t *models.Torrent, p *models.Peer) error {
func (tx *Tx) RemoveLeecher(t *storage.Torrent, p *storage.Peer) error {
err := tx.removePeer(p, LeechersPrefix)
if err != nil {
return err
}
delete(t.Leechers, models.PeerMapKey(p))
delete(t.Leechers, storage.PeerMapKey(p))
return nil
}
// LeecherFinished moves a peer's hashkey from a torrent's leecher set to the seeder set and updates the peer.
// This modifies the torrent argument, as well as the torrent's set and peer's hash in Redis.
// This function does not return an error if the peer doesn't exist or is not in the torrent's leecher set.
func (tx *Tx) LeecherFinished(torrent *models.Torrent, peer *models.Peer) error {
func (tx *Tx) LeecherFinished(torrent *storage.Torrent, peer *storage.Peer) error {
torrentIdKey := strconv.FormatUint(torrent.ID, 36)
seederSetKey := tx.conf.Prefix + SeedersPrefix + torrentIdKey
leecherSetKey := tx.conf.Prefix + LeechersPrefix + torrentIdKey
@ -606,8 +606,8 @@ func (tx *Tx) LeecherFinished(torrent *models.Torrent, peer *models.Peer) error
if err != nil {
return err
}
torrent.Seeders[models.PeerMapKey(peer)] = *peer
delete(torrent.Leechers, models.PeerMapKey(peer))
torrent.Seeders[storage.PeerMapKey(peer)] = *peer
delete(torrent.Leechers, storage.PeerMapKey(peer))
err = tx.setPeer(peer)
return err
@ -617,7 +617,7 @@ func (tx *Tx) LeecherFinished(torrent *models.Torrent, peer *models.Peer) error
// This modifies the torrent argument, as well as the torrent's set and peer's hash in Redis.
// This function does not return an error if the seeder already exists.
// This is a multiple action command, it's not internally atomic.
func (tx *Tx) AddSeeder(torrent *models.Torrent, peer *models.Peer) error {
func (tx *Tx) AddSeeder(torrent *storage.Torrent, peer *storage.Peer) error {
setKey := tx.conf.Prefix + SeedersPrefix + strconv.FormatUint(torrent.ID, 36)
_, err := tx.Do("SADD", setKey, getPeerHashKey(peer))
if err != nil {
@ -628,9 +628,9 @@ func (tx *Tx) AddSeeder(torrent *models.Torrent, peer *models.Peer) error {
return err
}
if torrent.Seeders == nil {
torrent.Seeders = make(map[string]models.Peer)
torrent.Seeders = make(map[string]storage.Peer)
}
torrent.Seeders[models.PeerMapKey(peer)] = *peer
torrent.Seeders[storage.PeerMapKey(peer)] = *peer
return nil
}
@ -638,30 +638,30 @@ func (tx *Tx) AddSeeder(torrent *models.Torrent, peer *models.Peer) error {
// This modifies the torrent argument, as well as the peer's hash in Redis.
// Setting assumes that the peer is already a seeder, and only needs to be updated.
// This function does not return an error if the seeder does not exist or is not in the torrent's seeder set.
func (tx *Tx) SetSeeder(t *models.Torrent, p *models.Peer) error {
func (tx *Tx) SetSeeder(t *storage.Torrent, p *storage.Peer) error {
err := tx.setPeer(p)
if err != nil {
return err
}
t.Seeders[models.PeerMapKey(p)] = *p
t.Seeders[storage.PeerMapKey(p)] = *p
return nil
}
// RemoveSeeder removes the given peer from a torrent's seeder set.
// This modifies the torrent argument, as well as the torrent's set and peer's hash in Redis.
// This function does not return an error if the peer doesn't exist, or is not in the set.
func (tx *Tx) RemoveSeeder(t *models.Torrent, p *models.Peer) error {
func (tx *Tx) RemoveSeeder(t *storage.Torrent, p *storage.Peer) error {
err := tx.removePeer(p, SeedersPrefix)
if err != nil {
return err
}
delete(t.Seeders, models.PeerMapKey(p))
delete(t.Seeders, storage.PeerMapKey(p))
return nil
}
// IncrementSlots increment a user's Slots by one.
// This function modifies the argument as well as the hash field in Redis.
func (tx *Tx) IncrementSlots(u *models.User) error {
func (tx *Tx) IncrementSlots(u *storage.User) error {
hashkey := tx.conf.Prefix + UserPrefix + u.Passkey
slotCount, err := redis.Int(tx.Do("HINCRBY", hashkey, "slots", 1))
if err != nil {
@ -673,7 +673,7 @@ func (tx *Tx) IncrementSlots(u *models.User) error {
// IncrementSlots increment a user's Slots by one.
// This function modifies the argument as well as the hash field in Redis.
func (tx *Tx) DecrementSlots(u *models.User) error {
func (tx *Tx) DecrementSlots(u *storage.User) error {
hashkey := tx.conf.Prefix + UserPrefix + u.Passkey
slotCount, err := redis.Int(tx.Do("HINCRBY", hashkey, "slots", -1))
if err != nil {
@ -685,5 +685,5 @@ func (tx *Tx) DecrementSlots(u *models.User) error {
// init registers the redis driver
func init() {
cache.Register("redis", &driver{})
tracker.Register("redis", &driver{})
}

0
cache/redis/redis_bench_test.go → storage/tracker/redis/redis_bench_test.go
View file

22
cache/redis/redis_test.go → storage/tracker/redis/redis_test.go
View file

@ -15,7 +15,7 @@ import (
"github.com/garyburd/redigo/redis"
"github.com/pushrax/chihaya/config"
"github.com/pushrax/chihaya/models"
"github.com/pushrax/chihaya/storage"
)
var (
@ -111,27 +111,27 @@ func createTestRedisTx() *Tx {
return txObj
}
func createTestUser() *models.User {
return &models.User{ID: createTestUserID(), Passkey: createTestPasskey(),
func createTestUser() *storage.User {
return &storage.User{ID: createTestUserID(), Passkey: createTestPasskey(),
UpMultiplier: 1.01, DownMultiplier: 1.0, Slots: 4, SlotsUsed: 2, Snatches: 7}
}
func createTestPeer(userID uint64, torrentID uint64) *models.Peer {
func createTestPeer(userID uint64, torrentID uint64) *storage.Peer {
return &models.Peer{ID: createTestPeerID(), UserID: userID, TorrentID: torrentID,
return &storage.Peer{ID: createTestPeerID(), UserID: userID, TorrentID: torrentID,
IP: "127.0.0.1", Port: 6889, Uploaded: 1024, Downloaded: 3000, Left: 4200, LastAnnounce: 11}
}
func createTestPeers(torrentID uint64, num int) map[string]models.Peer {
testPeers := make(map[string]models.Peer)
func createTestPeers(torrentID uint64, num int) map[string]storage.Peer {
testPeers := make(map[string]storage.Peer)
for i := 0; i < num; i++ {
tempPeer := createTestPeer(createTestUserID(), torrentID)
testPeers[models.PeerMapKey(tempPeer)] = *tempPeer
testPeers[storage.PeerMapKey(tempPeer)] = *tempPeer
}
return testPeers
}
func createTestTorrent() *models.Torrent {
func createTestTorrent() *storage.Torrent {
torrentInfohash := createTestInfohash()
torrentID := createTestTorrentID()
@ -139,7 +139,7 @@ func createTestTorrent() *models.Torrent {
testSeeders := createTestPeers(torrentID, 4)
testLeechers := createTestPeers(torrentID, 2)
testTorrent := models.Torrent{ID: torrentID, Infohash: torrentInfohash, Active: true,
testTorrent := storage.Torrent{ID: torrentID, Infohash: torrentInfohash, Active: true,
Seeders: testSeeders, Leechers: testLeechers, Snatches: 11, UpMultiplier: 1.0, DownMultiplier: 1.0, LastAction: 0}
return &testTorrent
}
@ -163,7 +163,7 @@ func TestInvalidPeers(t *testing.T) {
testTorrentID := createTestTorrentID()
testPeers := createTestPeers(testTorrentID, 3)
tempPeer := createTestPeer(createTestUserID(), testTorrentID)
testPeers[models.PeerMapKey(tempPeer)] = *tempPeer
testPeers[storage.PeerMapKey(tempPeer)] = *tempPeer
panicOnErr(testTx.addPeers(testPeers, "test:"))
// Imitate a peer being removed during get

52
cache/redis/tx_test.go → storage/tracker/redis/tx_test.go
View file

@ -11,17 +11,17 @@ import (
"testing"
"time"
"github.com/pushrax/chihaya/cache"
"github.com/pushrax/chihaya/config"
"github.com/pushrax/chihaya/models"
"github.com/pushrax/chihaya/storage"
"github.com/pushrax/chihaya/storage/tracker"
)
func createTestTx() cache.Tx {
func createTestTx() tracker.Conn {
testConfig, err := config.Open(os.Getenv("TESTCONFIGPATH"))
panicOnErr(err)
conf := &testConfig.Cache
testPool, err := cache.Open(conf)
testPool, err := tracker.Open(conf)
panicOnErr(err)
txObj, err := testPool.Get()
@ -216,11 +216,11 @@ func TestAddSeeder(t *testing.T) {
panicOnErr(tx.AddSeeder(testTorrent, testSeeder))
foundTorrent, found, err := tx.FindTorrent(testTorrent.Infohash)
panicOnErr(err)
foundSeeder, found := foundTorrent.Seeders[models.PeerMapKey(testSeeder)]
foundSeeder, found := foundTorrent.Seeders[storage.PeerMapKey(testSeeder)]
if found && foundSeeder != *testSeeder {
t.Error("seeder not added to cache", testSeeder)
}
foundSeeder, found = testTorrent.Seeders[models.PeerMapKey(testSeeder)]
foundSeeder, found = testTorrent.Seeders[storage.PeerMapKey(testSeeder)]
if found && foundSeeder != *testSeeder {
t.Error("seeder not added to local", testSeeder)
}
@ -237,11 +237,11 @@ func TestAddLeecher(t *testing.T) {
panicOnErr(tx.AddLeecher(testTorrent, testLeecher))
foundTorrent, found, err := tx.FindTorrent(testTorrent.Infohash)
panicOnErr(err)
foundLeecher, found := foundTorrent.Leechers[models.PeerMapKey(testLeecher)]
foundLeecher, found := foundTorrent.Leechers[storage.PeerMapKey(testLeecher)]
if found && foundLeecher != *testLeecher {
t.Error("leecher not added to cache", testLeecher)
}
foundLeecher, found = testTorrent.Leechers[models.PeerMapKey(testLeecher)]
foundLeecher, found = testTorrent.Leechers[storage.PeerMapKey(testLeecher)]
if found && foundLeecher != *testLeecher {
t.Error("leecher not added to local", testLeecher)
}
@ -257,14 +257,14 @@ func TestRemoveSeeder(t *testing.T) {
panicOnErr(tx.AddSeeder(testTorrent, testSeeder))
panicOnErr(tx.RemoveSeeder(testTorrent, testSeeder))
foundSeeder, found := testTorrent.Seeders[models.PeerMapKey(testSeeder)]
foundSeeder, found := testTorrent.Seeders[storage.PeerMapKey(testSeeder)]
if found || foundSeeder == *testSeeder {
t.Error("seeder not removed from local", foundSeeder)
}
foundTorrent, found, err := tx.FindTorrent(testTorrent.Infohash)
panicOnErr(err)
foundSeeder, found = foundTorrent.Seeders[models.PeerMapKey(testSeeder)]
foundSeeder, found = foundTorrent.Seeders[storage.PeerMapKey(testSeeder)]
if found || foundSeeder == *testSeeder {
t.Error("seeder not removed from cache", foundSeeder, *testSeeder)
}
@ -282,11 +282,11 @@ func TestRemoveLeecher(t *testing.T) {
panicOnErr(tx.RemoveLeecher(testTorrent, testLeecher))
foundTorrent, found, err := tx.FindTorrent(testTorrent.Infohash)
panicOnErr(err)
foundLeecher, found := foundTorrent.Leechers[models.PeerMapKey(testLeecher)]
foundLeecher, found := foundTorrent.Leechers[storage.PeerMapKey(testLeecher)]
if found || foundLeecher == *testLeecher {
t.Error("leecher not removed from cache", foundLeecher, *testLeecher)
}
foundLeecher, found = testTorrent.Leechers[models.PeerMapKey(testLeecher)]
foundLeecher, found = testTorrent.Leechers[storage.PeerMapKey(testLeecher)]
if found || foundLeecher == *testLeecher {
t.Error("leecher not removed from local", foundLeecher, *testLeecher)
}
@ -308,11 +308,11 @@ func TestSetSeeder(t *testing.T) {
foundTorrent, _, err := tx.FindTorrent(testTorrent.Infohash)
panicOnErr(err)
foundSeeder, _ := foundTorrent.Seeders[models.PeerMapKey(testSeeder)]
foundSeeder, _ := foundTorrent.Seeders[storage.PeerMapKey(testSeeder)]
if foundSeeder != *testSeeder {
t.Error("seeder not updated in cache", foundSeeder, *testSeeder)
}
foundSeeder, _ = testTorrent.Seeders[models.PeerMapKey(testSeeder)]
foundSeeder, _ = testTorrent.Seeders[storage.PeerMapKey(testSeeder)]
if foundSeeder != *testSeeder {
t.Error("seeder not updated in local", foundSeeder, *testSeeder)
}
@ -333,11 +333,11 @@ func TestSetLeecher(t *testing.T) {
panicOnErr(tx.SetLeecher(testTorrent, testLeecher))
foundTorrent, _, err := tx.FindTorrent(testTorrent.Infohash)
panicOnErr(err)
foundLeecher, _ := foundTorrent.Leechers[models.PeerMapKey(testLeecher)]
foundLeecher, _ := foundTorrent.Leechers[storage.PeerMapKey(testLeecher)]
if foundLeecher != *testLeecher {
t.Error("leecher not updated in cache", testLeecher)
}
foundLeecher, _ = testTorrent.Leechers[models.PeerMapKey(testLeecher)]
foundLeecher, _ = testTorrent.Leechers[storage.PeerMapKey(testLeecher)]
if foundLeecher != *testLeecher {
t.Error("leecher not updated in local", testLeecher)
}
@ -397,19 +397,19 @@ func TestLeecherFinished(t *testing.T) {
foundTorrent, _, err := tx.FindTorrent(testTorrent.Infohash)
panicOnErr(err)
foundSeeder, _ := foundTorrent.Seeders[models.PeerMapKey(testLeecher)]
foundSeeder, _ := foundTorrent.Seeders[storage.PeerMapKey(testLeecher)]
if foundSeeder != *testLeecher {
t.Error("seeder not added to cache", foundSeeder, *testLeecher)
}
foundSeeder, _ = foundTorrent.Leechers[models.PeerMapKey(testLeecher)]
foundSeeder, _ = foundTorrent.Leechers[storage.PeerMapKey(testLeecher)]
if foundSeeder == *testLeecher {
t.Error("leecher not removed from cache", testLeecher)
}
foundSeeder, _ = testTorrent.Seeders[models.PeerMapKey(testLeecher)]
foundSeeder, _ = testTorrent.Seeders[storage.PeerMapKey(testLeecher)]
if foundSeeder != *testLeecher {
t.Error("seeder not added to local", testLeecher)
}
foundSeeder, _ = testTorrent.Leechers[models.PeerMapKey(testLeecher)]
foundSeeder, _ = testTorrent.Leechers[storage.PeerMapKey(testLeecher)]
if foundSeeder == *testLeecher {
t.Error("leecher not removed from local", testLeecher)
}
@ -433,10 +433,10 @@ func TestUpdatePeer(t *testing.T) {
panicOnErr(tx.RemoveSeeder(testTorrent, testSeeder))
foundTorrent, _, err := tx.FindTorrent(testTorrent.Infohash)
panicOnErr(err)
if seeder, exists := foundTorrent.Seeders[models.PeerMapKey(testSeeder)]; exists {
if seeder, exists := foundTorrent.Seeders[storage.PeerMapKey(testSeeder)]; exists {
t.Error("seeder not removed from cache", seeder)
}
if seeder, exists := testTorrent.Seeders[models.PeerMapKey(testSeeder)]; exists {
if seeder, exists := testTorrent.Seeders[storage.PeerMapKey(testSeeder)]; exists {
t.Error("seeder not removed from local", seeder)
}
// Cleanup
@ -520,11 +520,11 @@ func TestParallelSetSeeder(t *testing.T) {
foundTorrent, _, err := tx.FindTorrent(testTorrent.Infohash)
panicOnErr(err)
foundSeeder, _ := foundTorrent.Seeders[models.PeerMapKey(testSeeder)]
foundSeeder, _ := foundTorrent.Seeders[storage.PeerMapKey(testSeeder)]
if foundSeeder != *testSeeder {
t.Error("seeder not updated in cache", foundSeeder, *testSeeder)
}
foundSeeder, _ = testTorrent.Seeders[models.PeerMapKey(testSeeder)]
foundSeeder, _ = testTorrent.Seeders[storage.PeerMapKey(testSeeder)]
if foundSeeder != *testSeeder {
t.Error("seeder not updated in local", foundSeeder, *testSeeder)
}
@ -549,11 +549,11 @@ func TestParallelAddLeecher(t *testing.T) {
foundTorrent, found, err := tx.FindTorrent(testTorrent.Infohash)
panicOnErr(err)
foundLeecher, found := foundTorrent.Leechers[models.PeerMapKey(testLeecher)]
foundLeecher, found := foundTorrent.Leechers[storage.PeerMapKey(testLeecher)]
if found && foundLeecher != *testLeecher {
t.Error("leecher not added to cache", testLeecher)
}
foundLeecher, found = testTorrent.Leechers[models.PeerMapKey(testLeecher)]
foundLeecher, found = testTorrent.Leechers[storage.PeerMapKey(testLeecher)]
if found && foundLeecher != *testLeecher {
t.Error("leecher not added to local", testLeecher)
}

85
storage/tracker/tracker.go Normal file
View file

@ -0,0 +1,85 @@
// Copyright 2013 The Chihaya Authors. All rights reserved.
// Use of this source code is governed by the BSD 2-Clause license,
// which can be found in the LICENSE file.
// Package tracker provides a generic interface for manipulating a
// BitTorrent tracker's fast-moving, inconsistent data.
package tracker
import (
"fmt"
"github.com/pushrax/chihaya/config"
"github.com/pushrax/chihaya/storage"
)
var (
drivers = make(map[string]Driver)
)
type Driver interface {
New(*config.DataStore) Pool
}
// Register makes a database driver available by the provided name.
// If Register is called twice with the same name or if driver is nil,
// it panics.
func Register(name string, driver Driver) {
if driver == nil {
panic("tracker: Register driver is nil")
}
if _, dup := drivers[name]; dup {
panic("tracker: Register called twice for driver " + name)
}
drivers[name] = driver
}
// Open creates a pool of data store connections specified by a storage configuration.
func Open(conf *config.DataStore) (Pool, error) {
driver, ok := drivers[conf.Driver]
if !ok {
return nil, fmt.Errorf(
"tracker: unknown driver %q (forgotten import?)",
conf.Driver,
)
}
pool := driver.New(conf)
return pool, nil
}
// Pool represents a thread-safe pool of connections to the data store
// that can be used to safely within concurrent goroutines.
type Pool interface {
Close() error
Get() (Conn, error)
}
// Conn represents a connection to the data store that can be used
// to make atomic and non-atomic reads/writes.
type Conn interface {
// Reads
FindUser(passkey string) (*storage.User, bool, error)
FindTorrent(infohash string) (*storage.Torrent, bool, error)
ClientWhitelisted(peerID string) (bool, error)
// Writes
RecordSnatch(u *storage.User, t *storage.Torrent) error
MarkActive(t *storage.Torrent) error
AddLeecher(t *storage.Torrent, p *storage.Peer) error
AddSeeder(t *storage.Torrent, p *storage.Peer) error
RemoveLeecher(t *storage.Torrent, p *storage.Peer) error
RemoveSeeder(t *storage.Torrent, p *storage.Peer) error
SetLeecher(t *storage.Torrent, p *storage.Peer) error
SetSeeder(t *storage.Torrent, p *storage.Peer) error
IncrementSlots(u *storage.User) error
DecrementSlots(u *storage.User) error
LeecherFinished(t *storage.Torrent, p *storage.Peer) error
// Priming / Testing
AddTorrent(t *storage.Torrent) error
RemoveTorrent(t *storage.Torrent) error
AddUser(u *storage.User) error
RemoveUser(u *storage.User) error
WhitelistClient(peerID string) error
UnWhitelistClient(peerID string) error
}