summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorFelicitas Pojtinger <felicitas@pojtinger.com>2022-06-18 23:05:52 +0200
committerFelicitas Pojtinger <felicitas@pojtinger.com>2022-06-18 23:05:52 +0200
commit50ba16d4ebfd9ad4d9d989f6f31eb5bf030ed51c (patch)
tree6b6630d68ac3846e479a0d5f85ab0acdb73df24d
parentb40a71e31050ffa054e308fa4d8d975a6c74f867 (diff)
feat: Add embedding APIv0.2.0
-rw-r--r--README.md4
-rw-r--r--cmd/htorrent/cmd/gateway.go257
-rw-r--r--cmd/htorrent/cmd/info.go64
-rw-r--r--go.mod3
-rw-r--r--go.sum5
-rw-r--r--pkg/api/http/v1/info.go7
-rw-r--r--pkg/client/manager.go81
-rw-r--r--pkg/server/gateway.go309
8 files changed, 478 insertions, 252 deletions
diff --git a/README.md b/README.md
index 4fc8456..9e90bf3 100644
--- a/README.md
+++ b/README.md
@@ -113,7 +113,7 @@ $ sudo firewall-cmd --reload
It should now be reachable on [localhost:1337](http://localhost:1337/).
-To use it in production, put this gateway behind a TLS-enabled reverse proxy such as [Caddy](https://caddyserver.com/) or [Traefik](https://traefik.io/). For the best security, you should use OpenID Connect to authenticate; for more information, see the [gateway reference](#gateway).
+To use it in production, put this gateway behind a TLS-enabled reverse proxy such as [Caddy](https://caddyserver.com/) or [Traefik](https://traefik.io/). For the best security, you should use OpenID Connect to authenticate; for more information, see the [gateway reference](#gateway). You can also embed the gateway in your own application using it's [Go API](https://pkg.go.dev/github.com/pojntfx/htorrent/pkg/server).
### 2. Get Torrent Infos with `htorrent info`
@@ -152,7 +152,7 @@ $ curl -u "admin:${API_PASSWORD}" -L -G --data-urlencode 'magnet=magnet:?xt=urn:
[{"path":"Sintel/Sintel.de.srt","length":1652,"creationTime":1655501577},{"path":"Sintel/Sintel.en.srt","length":1514,"creationTime":1655501577},{"path":"Sintel/Sintel.es.srt","length":1554,"creationTime":1655501577},{"path":"Sintel/Sintel.fr.srt","length":1618,"creationTime":1655501577},{"path":"Sintel/Sintel.it.srt","length":1546,"creationTime":1655501577},{"path":"Sintel/Sintel.mp4","length":129241752,"creationTime":1655501577},{"path":"Sintel/Sintel.nl.srt","length":1537,"creationTime":1655501577},{"path":"Sintel/Sintel.pl.srt","length":1536,"creationTime":1655501577},{"path":"Sintel/Sintel.pt.srt","length":1551,"creationTime":1655501577},{"path":"Sintel/Sintel.ru.srt","length":2016,"creationTime":1655501577},{"path":"Sintel/poster.jpg","length":46115,"creationTime":1655501577}
```
-For more information, see the [info reference](#info).
+For more information, see the [info reference](#info). You can also embed the client in your own application using it's [Go API](https://pkg.go.dev/github.com/pojntfx/htorrent/pkg/client).
### 3. Stream using a Media Player or cURL with `htorrent info -x`
diff --git a/cmd/htorrent/cmd/gateway.go b/cmd/htorrent/cmd/gateway.go
index 0a4b24e..efab2da 100644
--- a/cmd/htorrent/cmd/gateway.go
+++ b/cmd/htorrent/cmd/gateway.go
@@ -2,21 +2,14 @@ package cmd
import (
"context"
- "encoding/json"
- "errors"
- "fmt"
- "net/http"
+ "net"
"os"
+ "os/signal"
"path/filepath"
- "strings"
- "time"
+ "strconv"
+ "syscall"
- "github.com/anacrolix/torrent"
- "github.com/anacrolix/torrent/storage"
- "github.com/pojntfx/go-auth-utils/pkg/authn"
- "github.com/pojntfx/go-auth-utils/pkg/authn/basic"
- "github.com/pojntfx/go-auth-utils/pkg/authn/oidc"
- "github.com/rs/zerolog"
+ "github.com/pojntfx/htorrent/pkg/server"
"github.com/rs/zerolog/log"
"github.com/spf13/cobra"
"github.com/spf13/viper"
@@ -31,51 +24,10 @@ const (
oidcClientIDFlag = "oidc-client-id"
)
-var (
- errEmptyMagnetLink = errors.New("could not work with empty magnet link")
- errEmptyPath = errors.New("could not work with empty path")
- errCouldNotFindPath = errors.New("could not find path in torrent")
-)
-
-type file struct {
- Path string `json:"path"`
- Length int64 `json:"length"`
- CreationDate int64 `json:"creationTime"`
-}
-
var gatewayCmd = &cobra.Command{
Use: "gateway",
Aliases: []string{"g"},
Short: "Start a gateway",
- PersistentPreRunE: func(cmd *cobra.Command, args []string) error {
- viper.SetEnvPrefix("")
- viper.SetEnvKeyReplacer(strings.NewReplacer("-", "_", ".", "_"))
-
- if err := viper.BindPFlags(cmd.PersistentFlags()); err != nil {
- return err
- }
-
- switch viper.GetInt(verboseFlag) {
- case 0:
- zerolog.SetGlobalLevel(zerolog.Disabled)
- case 1:
- zerolog.SetGlobalLevel(zerolog.PanicLevel)
- case 2:
- zerolog.SetGlobalLevel(zerolog.FatalLevel)
- case 3:
- zerolog.SetGlobalLevel(zerolog.ErrorLevel)
- case 4:
- zerolog.SetGlobalLevel(zerolog.WarnLevel)
- case 5:
- zerolog.SetGlobalLevel(zerolog.InfoLevel)
- case 6:
- zerolog.SetGlobalLevel(zerolog.DebugLevel)
- default:
- zerolog.SetGlobalLevel(zerolog.TraceLevel)
- }
-
- return nil
- },
RunE: func(cmd *cobra.Command, args []string) error {
if err := viper.BindPFlags(cmd.PersistentFlags()); err != nil {
return err
@@ -84,185 +36,74 @@ var gatewayCmd = &cobra.Command{
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
- cfg := torrent.NewDefaultClientConfig()
-
- if viper.GetInt(verboseFlag) > 5 {
- cfg.Debug = true
- }
-
- cfg.DefaultStorage = storage.NewFileByInfoHash(viper.GetString(storageFlag))
-
- c, err := torrent.NewClient(cfg)
+ addr, err := net.ResolveTCPAddr("tcp", viper.GetString(laddrFlag))
if err != nil {
return err
}
- defer c.Close()
-
- var auth authn.Authn
- if strings.TrimSpace(viper.GetString(oidcIssuerFlag)) == "" && strings.TrimSpace(viper.GetString(oidcClientIDFlag)) == "" {
- auth = basic.NewAuthn(viper.GetString(apiUsernameFlag), viper.GetString(apiPasswordFlag))
- } else {
- auth = oidc.NewAuthn(viper.GetString(oidcIssuerFlag), viper.GetString(oidcClientIDFlag))
- }
-
- if err := auth.Open(ctx); err != nil {
- return err
- }
-
- mux := http.NewServeMux()
- mux.HandleFunc("/info", func(w http.ResponseWriter, r *http.Request) {
- u, p, ok := r.BasicAuth()
- if err := auth.Validate(u, p); !ok || err != nil {
- w.WriteHeader(http.StatusUnauthorized)
+ if port := os.Getenv("PORT"); port != "" {
+ log.Debug().Msg("Using port from PORT env variable")
- panic(fmt.Errorf("%v", http.StatusUnauthorized))
- }
-
- magnetLink := r.URL.Query().Get("magnet")
- if magnetLink == "" {
- w.WriteHeader(http.StatusUnprocessableEntity)
-
- panic(errEmptyMagnetLink)
- }
-
- log.Info().
- Str("magnet", magnetLink).
- Msg("Getting info")
-
- t, err := c.AddMagnet(magnetLink)
+ p, err := strconv.Atoi(port)
if err != nil {
- panic(err)
+ return err
}
- <-t.GotInfo()
- files := []file{}
- for _, f := range t.Files() {
- log.Info().
- Str("magnet", magnetLink).
- Str("path", f.Path()).
- Msg("Got info")
-
- files = append(files, file{
- Path: f.Path(),
- Length: f.Length(),
- CreationDate: f.Torrent().Metainfo().CreationDate,
- })
- }
-
- enc := json.NewEncoder(w)
- if err := enc.Encode(files); err != nil {
- panic(err)
- }
- })
-
- mux.HandleFunc("/stream", func(w http.ResponseWriter, r *http.Request) {
- defer func() {
- err := recover()
-
- switch err {
- case nil:
- fallthrough
- case http.StatusUnauthorized:
- fallthrough
- case http.StatusUnprocessableEntity:
- fallthrough
- case http.StatusNotFound:
- fallthrough
- default:
- w.WriteHeader(http.StatusInternalServerError)
+ addr.Port = p
+ }
- log.Debug().
- Err(err.(error)).
- Msg("Closed connection for client")
- }
- }()
+ gateway := server.NewGateway(
+ addr.String(),
+ viper.GetString(storageFlag),
+ viper.GetString(apiUsernameFlag),
+ viper.GetString(apiPasswordFlag),
+ viper.GetString(oidcIssuerFlag),
+ viper.GetString(oidcClientIDFlag),
+ viper.GetInt(verboseFlag) > 5,
+ func(peers int, total, completed int64, path string) {
+ log.Debug().
+ Int("peers", peers).
+ Int64("total", total).
+ Int64("completed", completed).
+ Str("path", path).
+ Msg("Streaming")
+ },
+ ctx,
+ )
- u, p, ok := r.BasicAuth()
- if err := auth.Validate(u, p); !ok || err != nil {
- w.WriteHeader(http.StatusUnauthorized)
+ if err := gateway.Open(); err != nil {
+ return err
+ }
- panic(fmt.Errorf("%v", http.StatusUnauthorized))
- }
+ s := make(chan os.Signal)
+ signal.Notify(s, os.Interrupt, syscall.SIGTERM)
+ go func() {
+ <-s
- magnetLink := r.URL.Query().Get("magnet")
- if magnetLink == "" {
- w.WriteHeader(http.StatusUnprocessableEntity)
+ log.Debug().Msg("Gracefully shutting down")
- panic(errEmptyMagnetLink)
- }
+ go func() {
+ <-s
- path := r.URL.Query().Get("path")
- if path == "" {
- w.WriteHeader(http.StatusUnprocessableEntity)
+ log.Debug().Msg("Forcing shutdown")
- panic(errEmptyPath)
- }
+ cancel()
- log.Info().
- Str("magnet", magnetLink).
- Str("path", path).
- Msg("Getting stream")
+ os.Exit(1)
+ }()
- t, err := c.AddMagnet(magnetLink)
- if err != nil {
+ if err := gateway.Close(); err != nil {
panic(err)
}
- <-t.GotInfo()
-
- found := false
- for _, f := range t.Files() {
- if f.Path() != path {
- continue
- }
-
- found = true
-
- go func() {
- tick := time.NewTicker(time.Millisecond * 100)
- defer tick.Stop()
-
- lastCompleted := int64(0)
- for range tick.C {
- if completed, total := f.BytesCompleted(), f.Length(); completed < total {
- if completed != lastCompleted {
- log.Debug().
- Int("peers", len(f.Torrent().PeerConns())).
- Int64("completed", completed).
- Int64("total", total).
- Str("path", f.Path()).
- Msg("Streaming")
- }
-
- lastCompleted = completed
- } else {
- return
- }
- }
- }()
-
- log.Info().
- Str("magnet", magnetLink).
- Str("path", path).
- Msg("Got stream")
-
- http.ServeContent(w, r, f.DisplayPath(), time.Unix(f.Torrent().Metainfo().CreationDate, 0), f.NewReader())
- }
-
- if !found {
- w.WriteHeader(http.StatusNotFound)
-
- panic(errCouldNotFindPath)
- }
- c.WaitAll()
- })
+ cancel()
+ }()
log.Info().
- Str("address", viper.GetString(laddrFlag)).
+ Str("address", addr.String()).
Msg("Listening")
- return http.ListenAndServe(viper.GetString(laddrFlag), mux)
+ return gateway.Wait()
},
}
diff --git a/cmd/htorrent/cmd/info.go b/cmd/htorrent/cmd/info.go
index 42e0b02..7f8311d 100644
--- a/cmd/htorrent/cmd/info.go
+++ b/cmd/htorrent/cmd/info.go
@@ -1,17 +1,18 @@
package cmd
import (
+ "context"
"encoding/csv"
- "encoding/json"
"errors"
"fmt"
- "net/http"
"net/url"
"os"
"regexp"
"strings"
"time"
+ "github.com/pojntfx/htorrent/pkg/client"
+ "github.com/pojntfx/htorrent/pkg/server"
"github.com/spf13/cobra"
"github.com/spf13/viper"
)
@@ -46,49 +47,23 @@ var infoCmd = &cobra.Command{
}
if strings.TrimSpace(viper.GetString(magnetFlag)) == "" {
- return errEmptyMagnetLink
+ return server.ErrEmptyMagnetLink
}
- hc := &http.Client{}
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
- baseURL, err := url.Parse(viper.GetString(raddrFlag))
- if err != nil {
- return err
- }
-
- infoSuffix, err := url.Parse("/info")
- if err != nil {
- return err
- }
-
- info := baseURL.ResolveReference(infoSuffix)
-
- q := info.Query()
- q.Set("magnet", viper.GetString(magnetFlag))
- info.RawQuery = q.Encode()
-
- req, err := http.NewRequest(http.MethodGet, info.String(), http.NoBody)
- if err != nil {
- return err
- }
- req.SetBasicAuth(viper.GetString(apiUsernameFlag), viper.GetString(apiPasswordFlag))
+ manager := client.NewManager(
+ viper.GetString(raddrFlag),
+ viper.GetString(apiUsernameFlag),
+ viper.GetString(apiPasswordFlag),
+ ctx,
+ )
- res, err := hc.Do(req)
+ files, err := manager.GetInfo(viper.GetString(magnetFlag))
if err != nil {
return err
}
- if res.Body != nil {
- defer res.Body.Close()
- }
- if res.StatusCode != http.StatusOK {
- return errors.New(res.Status)
- }
-
- files := []file{}
- dec := json.NewDecoder(res.Body)
- if err := dec.Decode(&files); err != nil {
- return err
- }
if strings.TrimSpace(viper.GetString(expressionFlag)) == "" {
w := csv.NewWriter(os.Stdout)
@@ -99,7 +74,7 @@ var infoCmd = &cobra.Command{
}
for _, f := range files {
- streamURL, err := getStreamURL(baseURL, viper.GetString(magnetFlag), f.Path)
+ streamURL, err := getStreamURL(viper.GetString(raddrFlag), viper.GetString(magnetFlag), f.Path)
if err != nil {
return err
}
@@ -113,7 +88,7 @@ var infoCmd = &cobra.Command{
for _, f := range files {
if exp.Match([]byte(f.Path)) {
- streamURL, err := getStreamURL(baseURL, viper.GetString(magnetFlag), f.Path)
+ streamURL, err := getStreamURL(viper.GetString(raddrFlag), viper.GetString(magnetFlag), f.Path)
if err != nil {
return err
}
@@ -131,13 +106,18 @@ var infoCmd = &cobra.Command{
},
}
-func getStreamURL(base *url.URL, magnet, path string) (string, error) {
+func getStreamURL(base string, magnet, path string) (string, error) {
+ baseURL, err := url.Parse(base)
+ if err != nil {
+ return "", err
+ }
+
streamSuffix, err := url.Parse("/stream")
if err != nil {
return "", err
}
- stream := base.ResolveReference(streamSuffix)
+ stream := baseURL.ResolveReference(streamSuffix)
q := stream.Query()
q.Set("magnet", magnet)
diff --git a/go.mod b/go.mod
index 087f605..91fef4e 100644
--- a/go.mod
+++ b/go.mod
@@ -4,6 +4,7 @@ go 1.18
require (
github.com/anacrolix/torrent v1.44.0
+ github.com/json-iterator/go v1.1.12
github.com/pojntfx/go-auth-utils v0.1.0
github.com/rs/zerolog v1.27.0
github.com/spf13/cobra v1.4.0
@@ -50,6 +51,8 @@ require (
github.com/mattn/go-colorable v0.1.12 // indirect
github.com/mattn/go-isatty v0.0.14 // indirect
github.com/mitchellh/mapstructure v1.5.0 // indirect
+ github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
+ github.com/modern-go/reflect2 v1.0.2 // indirect
github.com/mschoch/smat v0.2.0 // indirect
github.com/pelletier/go-toml v1.9.5 // indirect
github.com/pelletier/go-toml/v2 v2.0.1 // indirect
diff --git a/go.sum b/go.sum
index 852138a..d3337bb 100644
--- a/go.sum
+++ b/go.sum
@@ -275,6 +275,8 @@ github.com/inconshreveable/mousetrap v1.0.0 h1:Z8tu5sraLXCXIcARxBp/8cbvlwVa7Z1NH
github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANytuPF1OarO4DADm73n8=
github.com/json-iterator/go v1.1.6/go.mod h1:+SdeFBvtyEkXs7REEP0seUULqWtbJapLOCVDaaPEHmU=
github.com/json-iterator/go v1.1.9/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4=
+github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM=
+github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo=
github.com/jstemmer/go-junit-report v0.0.0-20190106144839-af01ea7f8024/go.mod h1:6v2b51hI/fHJwM22ozAgKL4VKDeJcHhJFhtBdhmNjmU=
github.com/jstemmer/go-junit-report v0.9.1/go.mod h1:Brl9GWCQeLvo8nXZwPNNblvFj/XSXhF0NWZEnDohbsk=
github.com/jtolds/gls v4.2.1+incompatible/go.mod h1:QJZ7F/aHp+rZTRtaJ1ow/lLfFfVYBRgL+9YlvaHOwJU=
@@ -304,9 +306,12 @@ github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5
github.com/mitchellh/mapstructure v1.5.0 h1:jeMsZIYE/09sWLaz43PL7Gy6RuMjD2eJVyuac5Z2hdY=
github.com/mitchellh/mapstructure v1.5.0/go.mod h1:bFUtVrKA4DC2yAKiSyO/QUcy7e+RRV2QTWOzhPopBRo=
github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
+github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w8PVh93nsPXa1VrQ6jlwL5oN8l14QlcNfg=
github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
github.com/modern-go/reflect2 v0.0.0-20180701023420-4b7aa43c6742/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0=
github.com/modern-go/reflect2 v1.0.1/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0=
+github.com/modern-go/reflect2 v1.0.2 h1:xBagoLtFs94CBntxluKeaWgTMpvLxC4ur3nMaC9Gz0M=
+github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk=
github.com/mschoch/smat v0.0.0-20160514031455-90eadee771ae/go.mod h1:qAyveg+e4CE+eKJXWVjKXM4ck2QobLqTDytGJbLLhJg=
github.com/mschoch/smat v0.2.0 h1:8imxQsjDm8yFEAVBe7azKmKSgzSkZXDuKkSq9374khM=
github.com/mschoch/smat v0.2.0/go.mod h1:kc9mz7DoBKqDyiRL7VZN8KvXQMWeTaVnttLRXOlotKw=
diff --git a/pkg/api/http/v1/info.go b/pkg/api/http/v1/info.go
new file mode 100644
index 0000000..a5c689a
--- /dev/null
+++ b/pkg/api/http/v1/info.go
@@ -0,0 +1,7 @@
+package v1
+
+type File struct {
+ Path string `json:"path"`
+ Length int64 `json:"length"`
+ CreationDate int64 `json:"creationTime"`
+}
diff --git a/pkg/client/manager.go b/pkg/client/manager.go
new file mode 100644
index 0000000..2ef7c03
--- /dev/null
+++ b/pkg/client/manager.go
@@ -0,0 +1,81 @@
+package client
+
+import (
+ "context"
+ "errors"
+ "net/http"
+ "net/url"
+
+ jsoniter "github.com/json-iterator/go"
+ v1 "github.com/pojntfx/htorrent/pkg/api/http/v1"
+)
+
+var (
+ json = jsoniter.ConfigCompatibleWithStandardLibrary
+)
+
+type Manager struct {
+ url string
+ username string
+ password string
+ ctx context.Context
+}
+
+func NewManager(
+ url string,
+ username string,
+ password string,
+ ctx context.Context,
+) *Manager {
+ return &Manager{
+ url: url,
+ username: username,
+ password: password,
+ ctx: ctx,
+ }
+}
+
+func (m *Manager) GetInfo(magnetLink string) ([]v1.File, error) {
+ hc := &http.Client{}
+
+ baseURL, err := url.Parse(m.url)
+ if err != nil {
+ return nil, err
+ }
+
+ infoSuffix, err := url.Parse("/info")
+ if err != nil {
+ return nil, err
+ }
+
+ info := baseURL.ResolveReference(infoSuffix)
+
+ q := info.Query()
+ q.Set("magnet", magnetLink)
+ info.RawQuery = q.Encode()
+
+ req, err := http.NewRequest(http.MethodGet, info.String(), http.NoBody)
+ if err != nil {
+ return nil, err
+ }
+ req.SetBasicAuth(m.username, m.password)
+
+ res, err := hc.Do(req)
+ if err != nil {
+ return nil, err
+ }
+ if res.Body != nil {
+ defer res.Body.Close()
+ }
+ if res.StatusCode != http.StatusOK {
+ return nil, errors.New(res.Status)
+ }
+
+ files := []v1.File{}
+ dec := json.NewDecoder(res.Body)
+ if err := dec.Decode(&files); err != nil {
+ return nil, err
+ }
+
+ return files, nil
+}
diff --git a/pkg/server/gateway.go b/pkg/server/gateway.go
new file mode 100644
index 0000000..b22100b
--- /dev/null
+++ b/pkg/server/gateway.go
@@ -0,0 +1,309 @@
+package server
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "net/http"
+ "strings"
+ "time"
+
+ "github.com/anacrolix/torrent"
+ "github.com/anacrolix/torrent/storage"
+ "github.com/pojntfx/go-auth-utils/pkg/authn"
+ "github.com/pojntfx/go-auth-utils/pkg/authn/basic"
+ "github.com/pojntfx/go-auth-utils/pkg/authn/oidc"
+ v1 "github.com/pojntfx/htorrent/pkg/api/http/v1"
+ "github.com/rs/zerolog/log"
+)
+
+var (
+ ErrEmptyMagnetLink = errors.New("could not work with empty magnet link")
+ ErrEmptyPath = errors.New("could not work with empty path")
+ ErrCouldNotFindPath = errors.New("could not find path in torrent")
+)
+
+type Gateway struct {
+ laddr string
+ storage string
+ apiUsername string
+ apiPassword string
+ oidcIssuer string
+ oidcClientID string
+ debug bool
+
+ onDownloadProgress func(peers int, total, completed int64, path string)
+
+ torrentClient *torrent.Client
+ srv *http.Server
+
+ errs chan error
+
+ ctx context.Context
+}
+
+func NewGateway(
+ laddr string,
+ storage string,
+ apiUsername string,
+ apiPassword string,
+ oidcIssuer string,
+ oidcClientID string,
+ debug bool,
+
+ onDownloadProgress func(peers int, total, completed int64, path string),
+
+ ctx context.Context,
+) *Gateway {
+ return &Gateway{
+ laddr: laddr,
+ storage: storage,
+ apiUsername: apiUsername,
+ apiPassword: apiPassword,
+ oidcIssuer: oidcIssuer,
+ oidcClientID: oidcClientID,
+ debug: debug,
+
+ onDownloadProgress: onDownloadProgress,
+
+ errs: make(chan error),
+
+ ctx: ctx,
+ }
+}
+
+func (g *Gateway) Open() error {
+ log.Trace().Msg("Opening gateway")
+
+ cfg := torrent.NewDefaultClientConfig()
+ cfg.Debug = g.debug
+ cfg.DefaultStorage = storage.NewFileByInfoHash(g.storage)
+
+ c, err := torrent.NewClient(cfg)
+ if err != nil {
+ return err
+ }
+ g.torrentClient = c
+
+ var auth authn.Authn
+ if strings.TrimSpace(g.oidcIssuer) == "" && strings.TrimSpace(g.oidcClientID) == "" {
+ auth = basic.NewAuthn(g.apiUsername, g.apiPassword)
+ } else {
+ auth = oidc.NewAuthn(g.oidcIssuer, g.oidcClientID)
+ }
+
+ if err := auth.Open(g.ctx); err != nil {
+ return err
+ }
+
+ mux := http.NewServeMux()
+
+ mux.HandleFunc("/info", func(w http.ResponseWriter, r *http.Request) {
+ u, p, ok := r.BasicAuth()
+ if err := auth.Validate(u, p); !ok || err != nil {
+ w.WriteHeader(http.StatusUnauthorized)
+
+ panic(fmt.Errorf("%v", http.StatusUnauthorized))
+ }
+
+ magnetLink := r.URL.Query().Get("magnet")
+ if magnetLink == "" {
+ w.WriteHeader(http.StatusUnprocessableEntity)
+
+ panic(ErrEmptyMagnetLink)
+ }
+
+ log.Debug().
+ Str("magnet", magnetLink).
+ Msg("Getting info")
+
+ t, err := c.AddMagnet(magnetLink)
+ if err != nil {
+ panic(err)
+ }
+ <-t.GotInfo()
+
+ files := []v1.File{}
+ for _, f := range t.Files() {
+ log.Debug().
+ Str("magnet", magnetLink).
+ Str("path", f.Path()).
+ Msg("Got info")
+
+ files = append(files, v1.File{
+ Path: f.Path(),
+ Length: f.Length(),
+ CreationDate: f.Torrent().Metainfo().CreationDate,
+ })
+ }
+
+ enc := json.NewEncoder(w)
+ if err := enc.Encode(files); err != nil {
+ panic(err)
+ }
+ })
+
+ mux.HandleFunc("/stream", func(w http.ResponseWriter, r *http.Request) {
+ defer func() {
+ err := recover()
+
+ switch err {
+ case nil:
+ fallthrough
+ case http.StatusUnauthorized:
+ fallthrough
+ case http.StatusUnprocessableEntity:
+ fallthrough
+ case http.StatusNotFound:
+ fallthrough
+ default:
+ w.WriteHeader(http.StatusInternalServerError)
+
+ log.Debug().
+ Err(err.(error)).
+ Msg("Closed connection for client")
+ }
+ }()
+
+ u, p, ok := r.BasicAuth()
+ if err := auth.Validate(u, p); !ok || err != nil {
+ w.WriteHeader(http.StatusUnauthorized)
+
+ panic(fmt.Errorf("%v", http.StatusUnauthorized))
+ }
+
+ magnetLink := r.URL.Query().Get("magnet")
+ if magnetLink == "" {
+ w.WriteHeader(http.StatusUnprocessableEntity)
+
+ panic(ErrEmptyMagnetLink)
+ }
+
+ path := r.URL.Query().Get("path")
+ if path == "" {
+ w.WriteHeader(http.StatusUnprocessableEntity)
+
+ panic(ErrEmptyPath)
+ }
+
+ log.Debug().
+ Str("magnet", magnetLink).
+ Str("path", path).
+ Msg("Getting stream")
+
+ t, err := c.AddMagnet(magnetLink)
+ if err != nil {
+ panic(err)
+ }
+ <-t.GotInfo()
+
+ found := false
+ for _, f := range t.Files() {
+ if f.Path() != path {
+ continue
+ }
+
+ found = true
+
+ go func() {
+ tick := time.NewTicker(time.Millisecond * 100)
+ defer tick.Stop()
+
+ lastCompleted := int64(0)
+ for range tick.C {
+ if completed, total := f.BytesCompleted(), f.Length(); completed < total {
+ if completed != lastCompleted {
+ log.Debug().
+ Int("peers", len(f.Torrent().PeerConns())).
+ Int64("total", total).
+ Int64("completed", completed).
+ Str("path", f.Path()).
+ Msg("Streaming")
+
+ g.onDownloadProgress(
+ len(f.Torrent().PeerConns()),
+ total,
+ completed,
+ f.Path(),
+ )
+ }
+
+ lastCompleted = completed
+ } else {
+ return
+ }
+ }
+ }()
+
+ log.Debug().
+ Str("magnet", magnetLink).
+ Str("path", path).
+ Msg("Got stream")
+
+ http.ServeContent(w, r, f.DisplayPath(), time.Unix(f.Torrent().Metainfo().CreationDate, 0), f.NewReader())
+ }
+
+ if !found {
+ w.WriteHeader(http.StatusNotFound)
+
+ panic(ErrCouldNotFindPath)
+ }
+
+ c.WaitAll()
+ })
+
+ g.srv = &http.Server{Addr: g.laddr}
+ g.srv.Handler = mux
+
+ log.Debug().
+ Str("address", g.laddr).
+ Msg("Listening")
+
+ go func() {
+ if err := g.srv.ListenAndServe(); err != nil {
+ if err == http.ErrServerClosed {
+ close(g.errs)
+
+ return
+ }
+
+ g.errs <- err
+
+ return
+ }
+ }()
+
+ return nil
+}
+
+func (g *Gateway) Close() error {
+ log.Trace().Msg("Closing gateway")
+
+ if err := g.srv.Shutdown(g.ctx); err != nil {
+ if err != context.Canceled {
+ return err
+ }
+ }
+
+ errs := g.torrentClient.Close()
+ for _, err := range errs {
+ if err != nil {
+ if err != context.Canceled {
+ return err
+ }
+ }
+ }
+
+ return nil
+}
+
+func (g *Gateway) Wait() error {
+ for err := range g.errs {
+ if err != nil {
+ return err
+ }
+ }
+
+ return nil
+}