Style: code style

adapter/provider/fetcher.go

@@ -0,0 +1,184 @@
+package provider
+
+import (
+	"bytes"
+	"crypto/md5"
+	"io/ioutil"
+	"os"
+	"path/filepath"
+	"time"
+
+	"github.com/Dreamacro/clash/log"
+)
+
+var (
+	fileMode os.FileMode = 0666
+	dirMode  os.FileMode = 0755
+)
+
+type parser = func([]byte) (interface{}, error)
+
+type fetcher struct {
+	name      string
+	vehicle   Vehicle
+	updatedAt *time.Time
+	ticker    *time.Ticker
+	done      chan struct{}
+	hash      [16]byte
+	parser    parser
+	onUpdate  func(interface{})
+}
+
+func (f *fetcher) Name() string {
+	return f.name
+}
+
+func (f *fetcher) VehicleType() VehicleType {
+	return f.vehicle.Type()
+}
+
+func (f *fetcher) Initial() (interface{}, error) {
+	var (
+		buf     []byte
+		err     error
+		isLocal bool
+	)
+	if stat, fErr := os.Stat(f.vehicle.Path()); fErr == nil {
+		buf, err = ioutil.ReadFile(f.vehicle.Path())
+		modTime := stat.ModTime()
+		f.updatedAt = &modTime
+		isLocal = true
+	} else {
+		buf, err = f.vehicle.Read()
+	}
+
+	if err != nil {
+		return nil, err
+	}
+
+	proxies, err := f.parser(buf)
+	if err != nil {
+		if !isLocal {
+			return nil, err
+		}
+
+		// parse local file error, fallback to remote
+		buf, err = f.vehicle.Read()
+		if err != nil {
+			return nil, err
+		}
+
+		proxies, err = f.parser(buf)
+		if err != nil {
+			return nil, err
+		}
+
+		isLocal = false
+	}
+
+	if f.vehicle.Type() != File && !isLocal {
+		if err := safeWrite(f.vehicle.Path(), buf); err != nil {
+			return nil, err
+		}
+	}
+
+	f.hash = md5.Sum(buf)
+
+	// pull proxies automatically
+	if f.ticker != nil {
+		go f.pullLoop()
+	}
+
+	return proxies, nil
+}
+
+func (f *fetcher) Update() (interface{}, bool, error) {
+	buf, err := f.vehicle.Read()
+	if err != nil {
+		return nil, false, err
+	}
+
+	now := time.Now()
+	hash := md5.Sum(buf)
+	if bytes.Equal(f.hash[:], hash[:]) {
+		f.updatedAt = &now
+		return nil, true, nil
+	}
+
+	proxies, err := f.parser(buf)
+	if err != nil {
+		return nil, false, err
+	}
+
+	if f.vehicle.Type() != File {
+		if err := safeWrite(f.vehicle.Path(), buf); err != nil {
+			return nil, false, err
+		}
+	}
+
+	f.updatedAt = &now
+	f.hash = hash
+
+	return proxies, false, nil
+}
+
+func (f *fetcher) Destroy() error {
+	if f.ticker != nil {
+		f.done <- struct{}{}
+	}
+	return nil
+}
+
+func (f *fetcher) pullLoop() {
+	for {
+		select {
+		case <-f.ticker.C:
+			elm, same, err := f.Update()
+			if err != nil {
+				log.Warnln("[Provider] %s pull error: %s", f.Name(), err.Error())
+				continue
+			}
+
+			if same {
+				log.Debugln("[Provider] %s's proxies doesn't change", f.Name())
+				continue
+			}
+
+			log.Infoln("[Provider] %s's proxies update", f.Name())
+			if f.onUpdate != nil {
+				f.onUpdate(elm)
+			}
+		case <-f.done:
+			f.ticker.Stop()
+			return
+		}
+	}
+}
+
+func safeWrite(path string, buf []byte) error {
+	dir := filepath.Dir(path)
+
+	if _, err := os.Stat(dir); os.IsNotExist(err) {
+		if err := os.MkdirAll(dir, dirMode); err != nil {
+			return err
+		}
+	}
+
+	return ioutil.WriteFile(path, buf, fileMode)
+}
+
+func newFetcher(name string, interval time.Duration, vehicle Vehicle, parser parser, onUpdate func(interface{})) *fetcher {
+	var ticker *time.Ticker
+	if interval != 0 {
+		ticker = time.NewTicker(interval)
+	}
+
+	return &fetcher{
+		name:     name,
+		ticker:   ticker,
+		vehicle:  vehicle,
+		parser:   parser,
+		done:     make(chan struct{}, 1),
+		onUpdate: onUpdate,
+	}
+}

adapter/provider/healthcheck.go

@@ -0,0 +1,91 @@
+package provider
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	C "github.com/Dreamacro/clash/constant"
+
+	"go.uber.org/atomic"
+)
+
+const (
+	defaultURLTestTimeout = time.Second * 5
+)
+
+type HealthCheckOption struct {
+	URL      string
+	Interval uint
+}
+
+type HealthCheck struct {
+	url       string
+	proxies   []C.Proxy
+	interval  uint
+	lazy      bool
+	lastTouch *atomic.Int64
+	done      chan struct{}
+}
+
+func (hc *HealthCheck) process() {
+	ticker := time.NewTicker(time.Duration(hc.interval) * time.Second)
+
+	go hc.check()
+	for {
+		select {
+		case <-ticker.C:
+			now := time.Now().Unix()
+			if !hc.lazy || now-hc.lastTouch.Load() < int64(hc.interval) {
+				hc.check()
+			}
+		case <-hc.done:
+			ticker.Stop()
+			return
+		}
+	}
+}
+
+func (hc *HealthCheck) setProxy(proxies []C.Proxy) {
+	hc.proxies = proxies
+}
+
+func (hc *HealthCheck) auto() bool {
+	return hc.interval != 0
+}
+
+func (hc *HealthCheck) touch() {
+	hc.lastTouch.Store(time.Now().Unix())
+}
+
+func (hc *HealthCheck) check() {
+	ctx, cancel := context.WithTimeout(context.Background(), defaultURLTestTimeout)
+	wg := &sync.WaitGroup{}
+
+	for _, proxy := range hc.proxies {
+		wg.Add(1)
+
+		go func(p C.Proxy) {
+			p.URLTest(ctx, hc.url)
+			wg.Done()
+		}(proxy)
+	}
+
+	wg.Wait()
+	cancel()
+}
+
+func (hc *HealthCheck) close() {
+	hc.done <- struct{}{}
+}
+
+func NewHealthCheck(proxies []C.Proxy, url string, interval uint, lazy bool) *HealthCheck {
+	return &HealthCheck{
+		proxies:   proxies,
+		url:       url,
+		interval:  interval,
+		lazy:      lazy,
+		lastTouch: atomic.NewInt64(0),
+		done:      make(chan struct{}, 1),
+	}
+}

adapter/provider/parser.go

@@ -0,0 +1,63 @@
+package provider
+
+import (
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/Dreamacro/clash/common/structure"
+	C "github.com/Dreamacro/clash/constant"
+)
+
+var (
+	errVehicleType = errors.New("unsupport vehicle type")
+)
+
+type healthCheckSchema struct {
+	Enable   bool   `provider:"enable"`
+	URL      string `provider:"url"`
+	Interval int    `provider:"interval"`
+	Lazy     bool   `provider:"lazy,omitempty"`
+}
+
+type proxyProviderSchema struct {
+	Type        string            `provider:"type"`
+	Path        string            `provider:"path"`
+	URL         string            `provider:"url,omitempty"`
+	Interval    int               `provider:"interval,omitempty"`
+	HealthCheck healthCheckSchema `provider:"health-check,omitempty"`
+}
+
+func ParseProxyProvider(name string, mapping map[string]interface{}) (ProxyProvider, error) {
+	decoder := structure.NewDecoder(structure.Option{TagName: "provider", WeaklyTypedInput: true})
+
+	schema := &proxyProviderSchema{
+		HealthCheck: healthCheckSchema{
+			Lazy: true,
+		},
+	}
+	if err := decoder.Decode(mapping, schema); err != nil {
+		return nil, err
+	}
+
+	var hcInterval uint
+	if schema.HealthCheck.Enable {
+		hcInterval = uint(schema.HealthCheck.Interval)
+	}
+	hc := NewHealthCheck([]C.Proxy{}, schema.HealthCheck.URL, hcInterval, schema.HealthCheck.Lazy)
+
+	path := C.Path.Resolve(schema.Path)
+
+	var vehicle Vehicle
+	switch schema.Type {
+	case "file":
+		vehicle = NewFileVehicle(path)
+	case "http":
+		vehicle = NewHTTPVehicle(schema.URL, path)
+	default:
+		return nil, fmt.Errorf("%w: %s", errVehicleType, schema.Type)
+	}
+
+	interval := time.Duration(uint(schema.Interval)) * time.Second
+	return NewProxySetProvider(name, interval, vehicle, hc), nil
+}

adapter/provider/provider.go

@@ -0,0 +1,261 @@
+package provider
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"runtime"
+	"time"
+
+	"github.com/Dreamacro/clash/adapter"
+	C "github.com/Dreamacro/clash/constant"
+
+	"gopkg.in/yaml.v2"
+)
+
+const (
+	ReservedName = "default"
+)
+
+// Provider Type
+const (
+	Proxy ProviderType = iota
+	Rule
+)
+
+// ProviderType defined
+type ProviderType int
+
+func (pt ProviderType) String() string {
+	switch pt {
+	case Proxy:
+		return "Proxy"
+	case Rule:
+		return "Rule"
+	default:
+		return "Unknown"
+	}
+}
+
+// Provider interface
+type Provider interface {
+	Name() string
+	VehicleType() VehicleType
+	Type() ProviderType
+	Initial() error
+	Update() error
+}
+
+// ProxyProvider interface
+type ProxyProvider interface {
+	Provider
+	Proxies() []C.Proxy
+	// ProxiesWithTouch is used to inform the provider that the proxy is actually being used while getting the list of proxies.
+	// Commonly used in Dial and DialUDP
+	ProxiesWithTouch() []C.Proxy
+	HealthCheck()
+}
+
+type ProxySchema struct {
+	Proxies []map[string]interface{} `yaml:"proxies"`
+}
+
+// for auto gc
+type ProxySetProvider struct {
+	*proxySetProvider
+}
+
+type proxySetProvider struct {
+	*fetcher
+	proxies     []C.Proxy
+	healthCheck *HealthCheck
+}
+
+func (pp *proxySetProvider) MarshalJSON() ([]byte, error) {
+	return json.Marshal(map[string]interface{}{
+		"name":        pp.Name(),
+		"type":        pp.Type().String(),
+		"vehicleType": pp.VehicleType().String(),
+		"proxies":     pp.Proxies(),
+		"updatedAt":   pp.updatedAt,
+	})
+}
+
+func (pp *proxySetProvider) Name() string {
+	return pp.name
+}
+
+func (pp *proxySetProvider) HealthCheck() {
+	pp.healthCheck.check()
+}
+
+func (pp *proxySetProvider) Update() error {
+	elm, same, err := pp.fetcher.Update()
+	if err == nil && !same {
+		pp.onUpdate(elm)
+	}
+	return err
+}
+
+func (pp *proxySetProvider) Initial() error {
+	elm, err := pp.fetcher.Initial()
+	if err != nil {
+		return err
+	}
+
+	pp.onUpdate(elm)
+	return nil
+}
+
+func (pp *proxySetProvider) Type() ProviderType {
+	return Proxy
+}
+
+func (pp *proxySetProvider) Proxies() []C.Proxy {
+	return pp.proxies
+}
+
+func (pp *proxySetProvider) ProxiesWithTouch() []C.Proxy {
+	pp.healthCheck.touch()
+	return pp.Proxies()
+}
+
+func proxiesParse(buf []byte) (interface{}, error) {
+	schema := &ProxySchema{}
+
+	if err := yaml.Unmarshal(buf, schema); err != nil {
+		return nil, err
+	}
+
+	if schema.Proxies == nil {
+		return nil, errors.New("file must have a `proxies` field")
+	}
+
+	proxies := []C.Proxy{}
+	for idx, mapping := range schema.Proxies {
+		proxy, err := adapter.ParseProxy(mapping)
+		if err != nil {
+			return nil, fmt.Errorf("proxy %d error: %w", idx, err)
+		}
+		proxies = append(proxies, proxy)
+	}
+
+	if len(proxies) == 0 {
+		return nil, errors.New("file doesn't have any valid proxy")
+	}
+
+	return proxies, nil
+}
+
+func (pp *proxySetProvider) setProxies(proxies []C.Proxy) {
+	pp.proxies = proxies
+	pp.healthCheck.setProxy(proxies)
+	if pp.healthCheck.auto() {
+		go pp.healthCheck.check()
+	}
+}
+
+func stopProxyProvider(pd *ProxySetProvider) {
+	pd.healthCheck.close()
+	pd.fetcher.Destroy()
+}
+
+func NewProxySetProvider(name string, interval time.Duration, vehicle Vehicle, hc *HealthCheck) *ProxySetProvider {
+	if hc.auto() {
+		go hc.process()
+	}
+
+	pd := &proxySetProvider{
+		proxies:     []C.Proxy{},
+		healthCheck: hc,
+	}
+
+	onUpdate := func(elm interface{}) {
+		ret := elm.([]C.Proxy)
+		pd.setProxies(ret)
+	}
+
+	fetcher := newFetcher(name, interval, vehicle, proxiesParse, onUpdate)
+	pd.fetcher = fetcher
+
+	wrapper := &ProxySetProvider{pd}
+	runtime.SetFinalizer(wrapper, stopProxyProvider)
+	return wrapper
+}
+
+// for auto gc
+type CompatibleProvider struct {
+	*compatibleProvider
+}
+
+type compatibleProvider struct {
+	name        string
+	healthCheck *HealthCheck
+	proxies     []C.Proxy
+}
+
+func (cp *compatibleProvider) MarshalJSON() ([]byte, error) {
+	return json.Marshal(map[string]interface{}{
+		"name":        cp.Name(),
+		"type":        cp.Type().String(),
+		"vehicleType": cp.VehicleType().String(),
+		"proxies":     cp.Proxies(),
+	})
+}
+
+func (cp *compatibleProvider) Name() string {
+	return cp.name
+}
+
+func (cp *compatibleProvider) HealthCheck() {
+	cp.healthCheck.check()
+}
+
+func (cp *compatibleProvider) Update() error {
+	return nil
+}
+
+func (cp *compatibleProvider) Initial() error {
+	return nil
+}
+
+func (cp *compatibleProvider) VehicleType() VehicleType {
+	return Compatible
+}
+
+func (cp *compatibleProvider) Type() ProviderType {
+	return Proxy
+}
+
+func (cp *compatibleProvider) Proxies() []C.Proxy {
+	return cp.proxies
+}
+
+func (cp *compatibleProvider) ProxiesWithTouch() []C.Proxy {
+	cp.healthCheck.touch()
+	return cp.Proxies()
+}
+
+func stopCompatibleProvider(pd *CompatibleProvider) {
+	pd.healthCheck.close()
+}
+
+func NewCompatibleProvider(name string, proxies []C.Proxy, hc *HealthCheck) (*CompatibleProvider, error) {
+	if len(proxies) == 0 {
+		return nil, errors.New("Provider need one proxy at least")
+	}
+
+	if hc.auto() {
+		go hc.process()
+	}
+
+	pd := &compatibleProvider{
+		name:        name,
+		proxies:     proxies,
+		healthCheck: hc,
+	}
+
+	wrapper := &CompatibleProvider{pd}
+	runtime.SetFinalizer(wrapper, stopCompatibleProvider)
+	return wrapper, nil
+}

adapter/provider/vehicle.go

@@ -0,0 +1,122 @@
+package provider
+
+import (
+	"context"
+	"io/ioutil"
+	"net/http"
+	"net/url"
+	"time"
+
+	"github.com/Dreamacro/clash/component/dialer"
+)
+
+// Vehicle Type
+const (
+	File VehicleType = iota
+	HTTP
+	Compatible
+)
+
+// VehicleType defined
+type VehicleType int
+
+func (v VehicleType) String() string {
+	switch v {
+	case File:
+		return "File"
+	case HTTP:
+		return "HTTP"
+	case Compatible:
+		return "Compatible"
+	default:
+		return "Unknown"
+	}
+}
+
+type Vehicle interface {
+	Read() ([]byte, error)
+	Path() string
+	Type() VehicleType
+}
+
+type FileVehicle struct {
+	path string
+}
+
+func (f *FileVehicle) Type() VehicleType {
+	return File
+}
+
+func (f *FileVehicle) Path() string {
+	return f.path
+}
+
+func (f *FileVehicle) Read() ([]byte, error) {
+	return ioutil.ReadFile(f.path)
+}
+
+func NewFileVehicle(path string) *FileVehicle {
+	return &FileVehicle{path: path}
+}
+
+type HTTPVehicle struct {
+	url  string
+	path string
+}
+
+func (h *HTTPVehicle) Type() VehicleType {
+	return HTTP
+}
+
+func (h *HTTPVehicle) Path() string {
+	return h.path
+}
+
+func (h *HTTPVehicle) Read() ([]byte, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second*20)
+	defer cancel()
+
+	uri, err := url.Parse(h.url)
+	if err != nil {
+		return nil, err
+	}
+
+	req, err := http.NewRequest(http.MethodGet, uri.String(), nil)
+	if err != nil {
+		return nil, err
+	}
+
+	if user := uri.User; user != nil {
+		password, _ := user.Password()
+		req.SetBasicAuth(user.Username(), password)
+	}
+
+	req = req.WithContext(ctx)
+
+	transport := &http.Transport{
+		// from http.DefaultTransport
+		MaxIdleConns:          100,
+		IdleConnTimeout:       90 * time.Second,
+		TLSHandshakeTimeout:   10 * time.Second,
+		ExpectContinueTimeout: 1 * time.Second,
+		DialContext:           dialer.DialContext,
+	}
+
+	client := http.Client{Transport: transport}
+	resp, err := client.Do(req)
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+
+	buf, err := ioutil.ReadAll(resp.Body)
+	if err != nil {
+		return nil, err
+	}
+
+	return buf, nil
+}
+
+func NewHTTPVehicle(url string, path string) *HTTPVehicle {
+	return &HTTPVehicle{url, path}
+}