Mirror/mochi
1
0
Fork 0
mirror of https://github.com/sot-tech/mochi.git synced 2026-07-12 11:38:11 -07:00
Code Issues Packages Projects Releases Wiki Activity

transition to httprouter

Browse Source
This commit is contained in:
Jimmy Zelinskie
2014-07-01 21:40:29 -04:00
parent 9cb5b82dc7
commit 6d5290d85e
14 changed files with 353 additions and 503 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/serve_announce.go
-299
View File
@@ -1,299 +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 server
import (
"fmt"
"io"
"net/http"
"github.com/golang/glog"
"github.com/chihaya/chihaya/bencode"
"github.com/chihaya/chihaya/drivers/tracker"
"github.com/chihaya/chihaya/models"
)
func (s Server) serveAnnounce(w http.ResponseWriter, r *http.Request) {
announce, err := models.NewAnnounce(r, s.conf)
if err != nil {
fail(err, w, r)
return
}
conn, err := s.trackerPool.Get()
if err != nil {
fail(err, w, r)
return
}
err = conn.ClientWhitelisted(announce.ClientID())
if err != nil {
fail(err, w, r)
return
}
var user *models.User
if s.conf.Private {
user, err = conn.FindUser(announce.Passkey)
if err != nil {
fail(err, w, r)
return
}
}
torrent, err := conn.FindTorrent(announce.Infohash)
if err != nil {
fail(err, w, r)
return
}
peer := models.NewPeer(announce, user, torrent)
created, err := updateTorrent(conn, announce, peer, torrent)
if err != nil {
fail(err, w, r)
return
}
snatched, err := handleEvent(conn, announce, peer, user, torrent)
if err != nil {
fail(err, w, r)
return
}
if s.conf.Private {
delta := models.NewAnnounceDelta(announce, peer, user, torrent, created, snatched)
s.backendConn.RecordAnnounce(delta)
}
writeAnnounceResponse(w, announce, user, torrent)
w.(http.Flusher).Flush()
if s.conf.Private {
glog.V(5).Infof(
"announce: ip: %s user: %s torrent: %s",
announce.IP,
user.ID,
torrent.ID,
)
} else {
glog.V(5).Infof("announce: ip: %s torrent: %s", announce.IP, torrent.ID)
}
}
func updateTorrent(c tracker.Conn, a *models.Announce, p *models.Peer, t *models.Torrent) (created bool, err error) {
if !t.Active && a.Left == 0 {
err = c.MarkActive(t)
if err != nil {
return
}
}
switch {
case t.InSeederPool(p):
err = c.SetSeeder(t, p)
if err != nil {
return
}
case t.InLeecherPool(p):
err = c.SetLeecher(t, p)
if err != nil {
return
}
default:
if a.Left == 0 {
err = c.AddSeeder(t, p)
if err != nil {
return
}
} else {
err = c.AddLeecher(t, p)
if err != nil {
return
}
}
created = true
}
return
}
func handleEvent(c tracker.Conn, a *models.Announce, p *models.Peer, u *models.User, t *models.Torrent) (snatched bool, err error) {
switch {
case a.Event == "stopped" || a.Event == "paused":
if t.InSeederPool(p) {
err = c.RemoveSeeder(t, p)
if err != nil {
return
}
}
if t.InLeecherPool(p) {
err = c.RemoveLeecher(t, p)
if err != nil {
return
}
}
case a.Event == "completed":
err = c.IncrementSnatches(t)
if err != nil {
return
}
snatched = true
if t.InLeecherPool(p) {
err = tracker.LeecherFinished(c, t, p)
if err != nil {
return
}
}
case t.InLeecherPool(p) && a.Left == 0:
// A leecher completed but the event was never received
err = tracker.LeecherFinished(c, t, p)
if err != nil {
return
}
}
return
}
func writeAnnounceResponse(w io.Writer, a *models.Announce, u *models.User, t *models.Torrent) {
seedCount := len(t.Seeders)
leechCount := len(t.Leechers)
var peerCount int
if a.Left == 0 {
peerCount = minInt(a.NumWant, leechCount)
} else {
peerCount = minInt(a.NumWant, leechCount+seedCount-1)
}
bencoder := bencode.NewEncoder(w)
fmt.Fprintf(w, "d")
bencoder.Encode("complete")
bencoder.Encode(seedCount)
bencoder.Encode("incomplete")
bencoder.Encode(leechCount)
bencoder.Encode("interval")
bencoder.Encode(a.Config.Announce.Duration)
bencoder.Encode("min interval")
bencoder.Encode(a.Config.MinAnnounce.Duration)
if a.NumWant > 0 && a.Event != "stopped" && a.Event != "paused" {
if a.Compact {
writePeersCompact(w, a, u, t, peerCount)
} else {
writePeersList(w, a, u, t, peerCount)
}
}
fmt.Fprintf(w, "e")
}
func writePeersCompact(w io.Writer, a *models.Announce, u *models.User, t *models.Torrent, peerCount int) {
ipv4s, ipv6s := getPeers(a, u, t, peerCount)
if len(ipv4s) > 0 {
// 6 is the number of bytes that represents 1 compact IPv4 address.
fmt.Fprintf(w, "peers%d:", len(ipv4s)*6)
for _, peer := range ipv4s {
if ip := peer.IP.To4(); ip != nil {
w.Write(ip)
w.Write([]byte{byte(peer.Port >> 8), byte(peer.Port & 0xff)})
}
}
}
if len(ipv6s) > 0 {
// 18 is the number of bytes that represents 1 compact IPv6 address.
fmt.Fprintf(w, "peers6%d:", len(ipv6s)*18)
for _, peer := range ipv6s {
if ip := peer.IP.To16(); ip != nil {
w.Write(ip)
w.Write([]byte{byte(peer.Port >> 8), byte(peer.Port & 0xff)})
}
}
}
}
func getPeers(a *models.Announce, u *models.User, t *models.Torrent, peerCount int) (ipv4s, ipv6s []*models.Peer) {
if a.Left == 0 {
// If they're seeding, give them only leechers.
splitPeers(&ipv4s, &ipv6s, a, u, t.Leechers, peerCount)
} else {
// If they're leeching, prioritize giving them seeders.
count := splitPeers(&ipv4s, &ipv6s, a, u, t.Seeders, peerCount)
splitPeers(&ipv4s, &ipv6s, a, u, t.Leechers, peerCount-count)
}
return
}
func splitPeers(ipv4s, ipv6s *[]*models.Peer, a *models.Announce, u *models.User, peers map[string]models.Peer, peerCount int) (count int) {
for _, peer := range peers {
if count >= peerCount {
break
}
if a.Config.Private && peer.UserID == u.ID {
continue
}
if ip := peer.IP.To4(); len(ip) == 4 {
*ipv4s = append(*ipv4s, &peer)
} else if ip := peer.IP.To16(); len(ip) == 16 {
*ipv6s = append(*ipv6s, &peer)
}
count++
}
return
}
func writePeersList(w io.Writer, a *models.Announce, u *models.User, t *models.Torrent, peerCount int) {
bencoder := bencode.NewEncoder(w)
ipv4s, ipv6s := getPeers(a, u, t, peerCount)
bencoder.Encode("peers")
fmt.Fprintf(w, "l")
for _, peer := range ipv4s {
writePeerDict(w, peer)
}
for _, peer := range ipv6s {
writePeerDict(w, peer)
}
fmt.Fprintf(w, "e")
}
func writePeerDict(w io.Writer, peer *models.Peer) {
bencoder := bencode.NewEncoder(w)
fmt.Fprintf(w, "d")
bencoder.Encode("ip")
bencoder.Encode(peer.IP.String())
bencoder.Encode("peer id")
bencoder.Encode(peer.ID)
bencoder.Encode("port")
bencoder.Encode(peer.Port)
fmt.Fprintf(w, "e")
}
func minInt(a, b int) int {
if a < b {
return a
}
return b
}
server/serve_announce_test.go
-86
View File
@@ -1,86 +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 server
import (
"net"
"net/http"
"net/http/httptest"
"testing"
"github.com/chihaya/chihaya/config"
"github.com/chihaya/chihaya/drivers/backend"
_ "github.com/chihaya/chihaya/drivers/backend/mock"
"github.com/chihaya/chihaya/drivers/tracker"
_ "github.com/chihaya/chihaya/drivers/tracker/mock"
"github.com/chihaya/chihaya/models"
)
func TestAnnounce(t *testing.T) {
s, err := New(config.New())
if err != nil {
t.Error(err)
}
err = s.Prime(func(t tracker.Pool, b backend.Conn) (err error) {
conn, err := t.Get()
if err != nil {
return
}
err = conn.AddUser(&models.User{
ID: 1,
Passkey: "yby47f04riwpndba456rqxtmifenq5h6",
})
if err != nil {
return
}
err = conn.WhitelistClient("TR2820")
if err != nil {
return
}
torrent := &models.Torrent{
ID: 1,
Infohash: string([]byte{0x89, 0xd4, 0xbc, 0x52, 0x11, 0x16, 0xca, 0x1d, 0x42, 0xa2, 0xf3, 0x0d, 0x1f, 0x27, 0x4d, 0x94, 0xe4, 0x68, 0x1d, 0xaf}),
Seeders: make(map[string]models.Peer),
Leechers: make(map[string]models.Peer),
}
err = conn.AddTorrent(torrent)
if err != nil {
return
}
err = conn.AddLeecher(torrent, &models.Peer{
ID: "-TR2820-l71jtqkl898b",
UserID: 1,
TorrentID: torrent.ID,
IP: net.ParseIP("127.0.0.1"),
Port: 34000,
Left: 0,
})
return
})
if err != nil {
t.Error(err)
}
url := "http://localhost:6881/yby47f04riwpndba456rqxtmifenq5h6/announce?info_hash=%89%d4%bcR%11%16%ca%1dB%a2%f3%0d%1f%27M%94%e4h%1d%af&peer_id=-TR2820-l71jtqkl898b&port=51413&uploaded=0&downloaded=0&left=0&numwant=1&key=3c8e3319&compact=0&supportcrypto=1"
r, err := http.NewRequest("GET", url, nil)
if err != nil {
t.Error(err)
}
w := httptest.NewRecorder()
s.serveAnnounce(w, r)
if w.Body.String() != "d8:completei1e10:incompletei1e8:intervali1800e12:min intervali900e5:peersld2:ip9:127.0.0.17:peer id20:-TR2820-l71jtqkl898b4:porti34000eeee" {
t.Errorf("improper response from server:\n%s", w.Body.String())
}
}
server/serve_scrape.go
-91
View File
@@ -1,91 +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 server
import (
"fmt"
"io"
"net/http"
"strings"
"github.com/golang/glog"
"github.com/chihaya/chihaya/bencode"
"github.com/chihaya/chihaya/models"
)
func (s *Server) serveScrape(w http.ResponseWriter, r *http.Request) {
scrape, err := models.NewScrape(r, s.conf)
if err != nil {
fail(err, w, r)
return
}
conn, err := s.trackerPool.Get()
if err != nil {
fail(err, w, r)
}
var user *models.User
if s.conf.Private {
user, err = conn.FindUser(scrape.Passkey)
if err != nil {
fail(err, w, r)
return
}
}
var (
torrents []*models.Torrent
torrentIDs []string
)
for _, infohash := range scrape.Infohashes {
torrent, err := conn.FindTorrent(infohash)
if err != nil {
fail(err, w, r)
return
}
torrents = append(torrents, torrent)
torrentIDs = append(torrentIDs, string(torrent.ID))
}
bencoder := bencode.NewEncoder(w)
fmt.Fprintf(w, "d")
bencoder.Encode("files")
for _, torrent := range torrents {
writeTorrentStatus(w, torrent)
}
fmt.Fprintf(w, "e")
w.(http.Flusher).Flush()
if s.conf.Private {
glog.V(5).Infof(
"scrape: ip: %s user: %s torrents: %s",
r.RemoteAddr,
user.ID,
strings.Join(torrentIDs, ", "),
)
} else {
glog.V(5).Infof(
"scrape: ip: %s torrents: %s",
r.RemoteAddr,
strings.Join(torrentIDs, ", "),
)
}
}
func writeTorrentStatus(w io.Writer, t *models.Torrent) {
bencoder := bencode.NewEncoder(w)
bencoder.Encode(t.Infohash)
fmt.Fprintf(w, "d")
bencoder.Encode("complete")
bencoder.Encode(len(t.Seeders))
bencoder.Encode("downloaded")
bencoder.Encode(t.Snatches)
bencoder.Encode("incomplete")
bencoder.Encode(len(t.Leechers))
fmt.Fprintf(w, "e")
}
server/serve_stats.go
-39
View File
@@ -1,39 +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 server
import (
"encoding/json"
"net/http"
"sync/atomic"
"time"
"github.com/chihaya/chihaya/config"
)
type stats struct {
Uptime config.Duration `json:"uptime"`
RPM int64 `json:"req_per_min"`
}
func (s *Server) serveStats(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
stats, _ := json.Marshal(&stats{
config.Duration{time.Now().Sub(s.startTime)},
s.rpm,
})
length, _ := w.Write(stats)
w.Header().Set("Content-Length", string(length))
w.(http.Flusher).Flush()
}
func (s *Server) updateStats() {
for _ = range time.NewTicker(time.Minute).C {
s.rpm = s.deltaRequests
atomic.StoreInt64(&s.deltaRequests, 0)
}
}
server/serve_stats_test.go
-38
View File
@@ -1,38 +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 server
import (
"net/http"
"net/http/httptest"
"testing"
"github.com/chihaya/chihaya/config"
_ "github.com/chihaya/chihaya/drivers/backend/mock"
_ "github.com/chihaya/chihaya/drivers/tracker/mock"
)
func TestStats(t *testing.T) {
s, err := New(config.New())
if err != nil {
t.Error(err)
}
r, err := http.NewRequest("GET", "127.0.0.1:80/stats", nil)
if err != nil {
t.Error(err)
}
w := httptest.NewRecorder()
s.serveStats(w, r)
if w.Code != 200 {
t.Error("/stats did not return 200 OK")
}
if w.Header()["Content-Type"][0] != "application/json" {
t.Error("/stats did not return JSON")
}
}
server/server.go
-138
View File
@@ -1,138 +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 server implements a BitTorrent tracker
package server
import (
"errors"
"io"
"net"
"net/http"
"path"
"strconv"
"sync/atomic"
"time"
"github.com/etix/stoppableListener"
"github.com/golang/glog"
"github.com/chihaya/chihaya/config"
"github.com/chihaya/chihaya/drivers/backend"
"github.com/chihaya/chihaya/drivers/tracker"
)
// Server represents BitTorrent tracker server.
type Server struct {
conf *config.Config
// These are open connections/pools.
listener *stoppableListener.StoppableListener
trackerPool tracker.Pool
backendConn backend.Conn
// These are for collecting stats.
startTime time.Time
deltaRequests int64
rpm int64
http.Server
}
// New creates a new Server.
func New(conf *config.Config) (*Server, error) {
trackerPool, err := tracker.Open(&conf.Tracker)
if err != nil {
return nil, err
}
backendConn, err := backend.Open(&conf.Backend)
if err != nil {
return nil, err
}
s := &Server{
conf: conf,
trackerPool: trackerPool,
backendConn: backendConn,
Server: http.Server{
Addr: conf.Addr,
ReadTimeout: conf.ReadTimeout.Duration,
},
}
s.Server.Handler = s
return s, nil
}
// ListenAndServe starts listening and handling incoming HTTP requests.
func (s *Server) ListenAndServe() error {
l, err := net.Listen("tcp", s.Addr)
if err != nil {
return err
}
sl := stoppableListener.Handle(l)
s.listener = sl
s.startTime = time.Now()
go s.updateStats()
s.Serve(s.listener)
return nil
}
// Stop cleanly ends the handling of incoming HTTP requests.
func (s *Server) Stop() error {
// Wait for current requests to finish being handled.
s.listener.Stop <- true
err := s.trackerPool.Close()
if err != nil {
return err
}
err = s.backendConn.Close()
if err != nil {
return err
}
return s.listener.Close()
}
func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
defer atomic.AddInt64(&s.deltaRequests, 1)
r.Close = true
_, action := path.Split(r.URL.Path)
switch action {
case "announce":
s.serveAnnounce(w, r)
return
case "scrape":
s.serveScrape(w, r)
return
case "stats":
s.serveStats(w, r)
return
default:
fail(errors.New("unknown action"), w, r)
return
}
}
func fail(err error, w http.ResponseWriter, r *http.Request) {
errmsg := err.Error()
msg := "d14:failure reason" + strconv.Itoa(len(errmsg)) + ":" + errmsg + "e"
length, _ := io.WriteString(w, msg)
w.Header().Add("Content-Length", string(length))
w.(http.Flusher).Flush()
glog.V(5).Infof(
"failed request: ip: %s failure: %s",
r.RemoteAddr,
errmsg,
)
}
server/server_test.go
-18
View File
@@ -1,18 +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 server
import (
"github.com/chihaya/chihaya/drivers/backend"
"github.com/chihaya/chihaya/drivers/tracker"
)
// Primer represents a function that can prime drivers with data.
type Primer func(tracker.Pool, backend.Conn) error
// Prime executes a priming function on the server.
func (s *Server) Prime(p Primer) error {
return p(s.trackerPool, s.backendConn)
}