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

copyright updated; imports renamed; misc fixes

Browse Source
This commit is contained in:
Jimmy Zelinskie
2013-06-21 21:43:11 -04:00
parent eee2810da6
commit 1bc42063ab
9 changed files with 176 additions and 129 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/announce.go
+27 -28
View File
@@ -1,58 +1,59 @@
// 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 (
"bytes"
"errors"
"log"
"fmt"
"net/http"
"path"
"github.com/jzelinskie/chihaya/config"
"github.com/jzelinskie/chihaya/storage"
"github.com/pushrax/chihaya/config"
)
func (h *handler) serveAnnounce(w *http.ResponseWriter, r *http.Request) {
buf := h.bufferpool.Take()
defer h.bufferpool.Give(buf)
defer h.writeResponse(&w, r, buf)
user, err := validatePasskey(dir, h.storage)
func (h *handler) serveAnnounce(w http.ResponseWriter, r *http.Request) {
passkey, action := path.Split(r.URL.Path)
user, err := validatePasskey(passkey, h.storage)
if err != nil {
fail(err, buf)
fail(err, w)
return
}
pq, err := parseQuery(r.URL.RawQuery)
if err != nil {
fail(errors.New("Error parsing query"), buf)
fail(errors.New("Error parsing query"), w)
return
}
ip, err := determineIP(r, pq)
ip, err := pq.determineIP(r)
if err != nil {
fail(err, buf)
fail(err, w)
return
}
err := validateParsedQuery(pq)
err = validateParsedQuery(pq)
if err != nil {
fail(errors.New("Malformed request"), buf)
fail(errors.New("Malformed request"), w)
return
}
if !whitelisted(peerId, h.conf) {
fail(errors.New("Your client is not approved"), buf)
if !whitelisted(pq.params["peerId"], h.conf) {
fail(errors.New("Your client is not approved"), w)
return
}
torrent, exists, err := h.storage.FindTorrent(infohash)
torrent, exists, err := h.storage.FindTorrent(pq.params["infohash"])
if err != nil {
panic("server: failed to find torrent")
}
if !exists {
fail(errors.New("This torrent does not exist"), buf)
fail(errors.New("This torrent does not exist"), w)
return
}
if torrent.Status == 1 && left == 0 {
if left, _ := pq.getUint64("left"); torrent.Status == 1 && left == 0 {
err := h.storage.UnpruneTorrent(torrent)
if err != nil {
panic("server: failed to unprune torrent")
@@ -65,17 +66,15 @@ func (h *handler) serveAnnounce(w *http.ResponseWriter, r *http.Request) {
torrent.Status,
left,
),
buf,
w,
)
return
}
//go
// TODO
}
func whitelisted(peerId string, conf config.Config) bool {
// TODO Decide if whitelist should be in storage or config
}
func newPeer() {
func whitelisted(peerId string, conf *config.Config) bool {
// TODO
return false
}
server/query.go
+40 -8
View File
@@ -1,7 +1,13 @@
// 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 (
"errors"
"net/http"
"net/url"
"strconv"
)
@@ -11,11 +17,11 @@ type parsedQuery struct {
}
func (pq *parsedQuery) getUint64(key string) (uint64, bool) {
str, exists := pq[key]
str, exists := pq.params[key]
if !exists {
return 0, false
}
val, err := strconv.Uint64(str, 10, 64)
val, err := strconv.ParseUint(str, 10, 64)
if err != nil {
return 0, false
}
@@ -54,11 +60,11 @@ func parseQuery(query string) (*parsedQuery, error) {
keyStr, err := url.QueryUnescape(query[keyStart : keyEnd+1])
if err != nil {
return err
return nil, err
}
valStr, err := url.QueryUnescape(query[valStart : valEnd+1])
if err != nil {
return err
return nil, err
}
pq.params[keyStr] = valStr
@@ -67,7 +73,7 @@ func parseQuery(query string) (*parsedQuery, error) {
if hasInfohash {
// Multiple infohashes
if pq.infohashes == nil {
pq.infohashes = []string{firstInfoHash}
pq.infohashes = []string{firstInfohash}
}
pq.infohashes = append(pq.infohashes, valStr)
} else {
@@ -87,15 +93,15 @@ func parseQuery(query string) (*parsedQuery, error) {
valEnd = i
}
}
return
return pq, nil
}
func validateParsedQuery(pq *parsedQuery) error {
infohash, ok := pq["info_hash"]
infohash, ok := pq.params["info_hash"]
if infohash == "" {
return errors.New("infohash does not exist")
}
peerId, ok := pq["peer_id"]
peerId, ok := pq.params["peer_id"]
if peerId == "" {
return errors.New("peerId does not exist")
}
@@ -117,3 +123,29 @@ func validateParsedQuery(pq *parsedQuery) error {
}
return nil
}
func (pq *parsedQuery) determineIP(r *http.Request) (string, error) {
ip, ok := pq.params["ip"]
if !ok {
ip, ok = pq.params["ipv4"]
if !ok {
ips, ok := r.Header["X-Real-Ip"]
if ok && len(ips) > 0 {
ip = ips[0]
} else {
portIndex := len(r.RemoteAddr) - 1
for ; portIndex >= 0; portIndex-- {
if r.RemoteAddr[portIndex] == ':' {
break
}
}
if portIndex != -1 {
ip = r.RemoteAddr[0:portIndex]
} else {
return "", errors.New("Failed to parse IP address")
}
}
}
}
return ip, nil
}
server/server.go
+81 -74
View File
@@ -1,8 +1,13 @@
// 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 (
"bytes"
"errors"
"fmt"
"io"
"net"
"net/http"
"path"
@@ -10,61 +15,81 @@ import (
"sync"
"sync/atomic"
"github.com/jzelinskie/bufferpool"
"github.com/jzelinskie/chihaya/config"
"github.com/jzelinskie/chihaya/storage"
"github.com/pushrax/chihaya/config"
"github.com/pushrax/chihaya/storage"
)
type Server struct {
conf *config.Config
listener net.Listener
storage storage.Storage
terminated *bool
waitgroup *sync.WaitGroup
http.Server
listener *net.Listener
}
func New(conf *config.Config) {
return &Server{
Addr: conf.Addr,
Handler: newHandler(conf),
func New(conf *config.Config) (*Server, error) {
var (
wg sync.WaitGroup
terminated bool
)
store, err := storage.New(&conf.Storage)
if err != nil {
return nil, err
}
handler := &handler{
conf: conf,
storage: store,
terminated: &terminated,
waitgroup: &wg,
}
s := &Server{
conf: conf,
storage: store,
terminated: &terminated,
waitgroup: &wg,
}
s.Server.Addr = conf.Addr
s.Server.Handler = handler
return s, nil
}
func (s *Server) Start() error {
s.listener, err = net.Listen("tcp", config.Addr)
listener, err := net.Listen("tcp", s.conf.Addr)
if err != nil {
return err
}
s.Handler.terminated = false
*s.terminated = false
s.Serve(s.listener)
s.Handler.waitgroup.Wait()
s.Handler.storage.Shutdown()
s.waitgroup.Wait()
return nil
}
func (s *Server) Stop() error {
s.Handler.waitgroup.Wait()
s.Handler.terminated = true
return s.Handler.listener.Close()
*s.terminated = true
s.waitgroup.Wait()
err := s.storage.Shutdown()
if err != nil {
return err
}
return s.listener.Close()
}
type handler struct {
bufferpool *bufferpool.BufferPool
conf *config.Config
deltaRequests int64
storage *storage.Storage
terminated bool
waitgroup sync.WaitGroup
}
func newHandler(conf *config.Config) {
return &Handler{
bufferpool: bufferpool.New(conf.BufferPoolSize, 500),
conf: conf,
storage: storage.New(&conf.Storage),
}
storage storage.Storage
terminated *bool
waitgroup *sync.WaitGroup
}
func (h *handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
if h.terminated {
if *h.terminated {
return
}
@@ -72,46 +97,54 @@ func (h *handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
defer h.waitgroup.Done()
if r.URL.Path == "/stats" {
h.serveStats(&w, r)
h.serveStats(w, r)
return
}
dir, action := path.Split(requestPath)
passkey, action := path.Split(r.URL.Path)
switch action {
case "announce":
h.serveAnnounce(&w, r)
h.serveAnnounce(w, r)
return
case "scrape":
// TODO
h.serveScrape(&w, r)
h.serveScrape(w, r)
return
default:
buf := h.bufferpool.Take()
fail(errors.New("Unknown action"), buf)
h.writeResponse(&w, r, buf)
written := fail(errors.New("Unknown action"), w)
h.finalizeResponse(w, r, written)
return
}
}
func writeResponse(w *http.ResponseWriter, r *http.Request, buf *bytes.Buffer) {
func (h *handler) finalizeResponse(
w http.ResponseWriter,
r *http.Request,
written int,
) {
r.Close = true
w.Header().Add("Content-Type", "text/plain")
w.Header().Add("Connection", "close")
w.Header().Add("Content-Length", strconv.Itoa(buf.Len()))
w.Write(buf.Bytes())
w.Header().Add("Content-Length", strconv.Itoa(written))
w.(http.Flusher).Flush()
atomic.AddInt64(h.deltaRequests, 1)
atomic.AddInt64(&h.deltaRequests, 1)
}
func fail(err error, buf *bytes.Buffer) {
buf.WriteString("d14:failure reason")
buf.WriteString(strconv.Itoa(len(err)))
buf.WriteRune(':')
buf.WriteString(err)
buf.WriteRune('e')
func fail(err error, w http.ResponseWriter) int {
e := err.Error()
message := fmt.Sprintf(
"%s%s%s%s%s",
"d14:failure reason",
strconv.Itoa(len(e)),
':',
e,
'e',
)
written, _ := io.WriteString(w, message)
return written
}
func validatePasskey(dir string, s *storage.Storage) (storage.User, error) {
func validatePasskey(dir string, s storage.Storage) (*storage.User, error) {
if len(dir) != 34 {
return nil, errors.New("Your passkey is invalid")
}
@@ -127,29 +160,3 @@ func validatePasskey(dir string, s *storage.Storage) (storage.User, error) {
return user, nil
}
func determineIP(r *http.Request, pq *parsedQuery) (string, error) {
ip, ok := pq.params["ip"]
if !ok {
ip, ok = pq.params["ipv4"]
if !ok {
ips, ok := r.Header["X-Real-Ip"]
if ok && len(ips) > 0 {
ip = ips[0]
} else {
portIndex := len(r.RemoteAddr) - 1
for ; portIndex >= 0; portIndex-- {
if r.RemoteAddr[portIndex] == ':' {
break
}
}
if portIndex != -1 {
ip = r.RemoteAddr[0:portIndex]
} else {
return "", errors.New("Failed to parse IP address")
}
}
}
}
return &ip, nil
}