Improve: using travis ci

.gitignore

@@ -0,0 +1,15 @@
+# Binaries for programs and plugins
+*.exe
+*.exe~
+*.dll
+*.so
+*.dylib
+
+# Test binary, build with `go test -c`
+*.test
+
+# Output of the go coverage tool, specifically when used with LiteIDE
+*.out
+
+# dep
+vendor

.travis.yml

@@ -0,0 +1,31 @@
+language: go
+sudo: false
+go:
+  - "1.10"
+before_install:
+  - go get -u github.com/golang/dep/cmd/dep
+install:
+  - "$GOPATH/bin/dep ensure"
+env:
+  global:
+    - NAME=clash
+    - BINDIR=bin
+script:
+  - go test
+  - GOARCH=amd64 GOOS=linux go build -o $BINDIR/$NAME-linux
+  - GOARCH=amd64 GOOS=darwin go build -o $BINDIR/$NAME-macos
+  - chmod +x $BINDIR/$NAME-*
+  - gzip $BINDIR/$NAME-linux
+  - gzip $BINDIR/$NAME-macos
+deploy:
+  provider: releases
+  prerelease: true
+  skip_cleanup: true
+  api_key:
+    secure: dp1tc1h0er7aaAZ1hY0Xk/cUKwB0ifsAjg6e0M/Ad5NC87oucP6ESNFkDu0e9rUS1yB826/VnVGzNE/Z5zdjXVzPft+g5v5oRxzI4BKLhf07t9s+x8Z+3sApTxdsC5BvcN9x+5yRbpDLQ3biDPxSFu86j7m2pkEWw6XYNZO3/5y+RZXX7zu+d4MzTLUaA2kWl7KQAP0tEJNuw9ACDhpkw7LYbU/8q3E76prOTeme5/AT6Gxj7XhKUNP27lazhhqBSWM14ybPANqojNLEfMFHN/Eu2phYO07MuLTd4zuOIuw9y65kgvTFcHRlORjwUhnviXyA69obQejjgDI1WDOtU4PqpFaSLrxWtKI6k5VNWHARYggDm/wKl0WG7F0Kgio1KiGGhDg2yrbseXr/zBNaDhBtTFh6XJffqqwmgby1PXB6PWwfvWXooJMaQiFZczLWeMBl8v6XbSN6jtMTh/PQlKai6BcDd4LM8GQ7VHpSeff4qXEU4Vpnadjgs8VDPOHng6/HV+wDs8q2LrlMbnxLWxbCjOMUB6w7YnSrwH9owzKSoUs/531I4tTCRQIgipJtTK2b881/8osVjdMGS1mDXhBWO+OM0LCAdORJz+kN4PIkXXvKLt6jX74k6z4M3swFaqqtlTduN2Yy/ErsjguQO1VZfHmcpNssmJXI5QB9sxA=
+  file: bin/*
+  file_glob: true
+  on:
+    repo: Dreamacro/clash
+    branch: master
+    tags: true

Gopkg.lock

@@ -0,0 +1,84 @@
+# This file is autogenerated, do not edit; changes may be undone by the next 'dep ensure'.
+
+
+[[projects]]
+  branch = "master"
+  name = "github.com/Yawning/chacha20"
+  packages = ["."]
+  revision = "e3b1f968fc6397b51d963fee8ec8711a47bc0ce8"
+
+[[projects]]
+  name = "github.com/eapache/queue"
+  packages = ["."]
+  revision = "44cc805cf13205b55f69e14bcb69867d1ae92f98"
+  version = "v1.1.0"
+
+[[projects]]
+  name = "github.com/oschwald/geoip2-golang"
+  packages = ["."]
+  revision = "7118115686e16b77967cdbf55d1b944fe14ad312"
+  version = "v1.2.1"
+
+[[projects]]
+  name = "github.com/oschwald/maxminddb-golang"
+  packages = ["."]
+  revision = "c5bec84d1963260297932a1b7a1753c8420717a7"
+  version = "v1.3.0"
+
+[[projects]]
+  name = "github.com/riobard/go-shadowsocks2"
+  packages = [
+    "core",
+    "shadowaead",
+    "shadowstream",
+    "socks"
+  ]
+  revision = "8346403248229fc7e10d7a259de8e9352a9d8830"
+  version = "v0.1.0"
+
+[[projects]]
+  name = "github.com/sirupsen/logrus"
+  packages = ["."]
+  revision = "c155da19408a8799da419ed3eeb0cb5db0ad5dbc"
+  version = "v1.0.5"
+
+[[projects]]
+  branch = "master"
+  name = "golang.org/x/crypto"
+  packages = [
+    "chacha20poly1305",
+    "hkdf",
+    "internal/chacha20",
+    "poly1305",
+    "ssh/terminal"
+  ]
+  revision = "8ac0e0d97ce45cd83d1d7243c060cb8461dda5e9"
+
+[[projects]]
+  branch = "master"
+  name = "golang.org/x/sys"
+  packages = [
+    "cpu",
+    "unix",
+    "windows"
+  ]
+  revision = "9527bec2660bd847c050fda93a0f0c6dee0800bb"
+
+[[projects]]
+  name = "gopkg.in/eapache/channels.v1"
+  packages = ["."]
+  revision = "47238d5aae8c0fefd518ef2bee46290909cf8263"
+  version = "v1.1.0"
+
+[[projects]]
+  name = "gopkg.in/ini.v1"
+  packages = ["."]
+  revision = "06f5f3d67269ccec1fe5fe4134ba6e982984f7f5"
+  version = "v1.37.0"
+
+[solve-meta]
+  analyzer-name = "dep"
+  analyzer-version = 1
+  inputs-digest = "4297c505508c6cdd8c94fd4ef29bc6940c65fea81e125fcf871a316b6e671a71"
+  solver-name = "gps-cdcl"
+  solver-version = 1

Gopkg.toml

@@ -0,0 +1,50 @@
+# Gopkg.toml example
+#
+# Refer to https://github.com/golang/dep/blob/master/docs/Gopkg.toml.md
+# for detailed Gopkg.toml documentation.
+#
+# required = ["github.com/user/thing/cmd/thing"]
+# ignored = ["github.com/user/project/pkgX", "bitbucket.org/user/project/pkgA/pkgY"]
+#
+# [[constraint]]
+#   name = "github.com/user/project"
+#   version = "1.0.0"
+#
+# [[constraint]]
+#   name = "github.com/user/project2"
+#   branch = "dev"
+#   source = "github.com/myfork/project2"
+#
+# [[override]]
+#   name = "github.com/x/y"
+#   version = "2.4.0"
+#
+# [prune]
+#   non-go = false
+#   go-tests = true
+#   unused-packages = true
+
+
+[[constraint]]
+  name = "github.com/oschwald/geoip2-golang"
+  version = "1.2.1"
+
+[[constraint]]
+  name = "github.com/riobard/go-shadowsocks2"
+  version = "0.1.0"
+
+[[constraint]]
+  name = "github.com/sirupsen/logrus"
+  version = "1.0.5"
+
+[[constraint]]
+  name = "gopkg.in/eapache/channels.v1"
+  version = "1.1.0"
+
+[[constraint]]
+  name = "gopkg.in/ini.v1"
+  version = "1.37.0"
+
+[prune]
+  go-tests = true
+  unused-packages = true

LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2018 Dreamacro
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

README.md

@@ -1 +1,62 @@
-Removed
+# Clash
+
+A rule based proxy in Go.
+
+## Features
+
+- HTTP/HTTPS and SOCKS proxy
+- Surge like configuration
+- GeoIP rule support
+
+## Install
+
+You can build from source:
+
+```sh
+go get -u -v github.com/Dreamacro/clash
+```
+
+Pre-built binaries are available: [release](https://github.com/Dreamacro/clash/releases)
+
+Requires Go >= 1.10.
+
+## Daemon
+
+Unfortunately, there is no native elegant way to implement golang's daemon.
+
+So we can use third-party daemon tools like pm2, supervisor, and so on.
+
+In the case of pm2, we can start the daemon this way:
+
+```sh
+pm2 start clash
+```
+
+## Config
+
+Configuration file at `$HOME/.config/clash/config.ini`
+
+Below is a simple demo configuration file:
+
+```ini
+[General]
+port = 7890
+socks-port = 7891
+
+[Proxy]
+# name = ss, server, port, cipher, password
+# The types of cipher are consistent with go-shadowsocks2
+# support AEAD_AES_128_GCM AEAD_AES_192_GCM AEAD_AES_256_GCM AEAD_CHACHA20_POLY1305 AES-128-CTR AES-192-CTR AES-256-CTR AES-128-CFB AES-192-CFB AES-256-CFB CHACHA20-IETF XCHACHA20
+Proxy = ss, server, port, AEAD_CHACHA20_POLY1305, password
+
+[Rule]
+DOMAIN-SUFFIX,google.com,Proxy
+DOMAIN-KEYWORD,google,Proxy
+DOMAIN-SUFFIX,ad.com,REJECT
+GEOIP,CN,DIRECT
+FINAL,,Proxy # note: there is two ","
+```
+
+## TODO
+
+- [ ] Complementing the necessary rule operators

adapters/direct.go

@@ -0,0 +1,44 @@
+package adapters
+
+import (
+	"io"
+	"net"
+
+	C "github.com/Dreamacro/clash/constant"
+)
+
+// DirectAdapter is a directly connected adapter
+type DirectAdapter struct {
+	conn net.Conn
+}
+
+// ReadWriter is used to handle network traffic
+func (d *DirectAdapter) ReadWriter() io.ReadWriter {
+	return d.conn
+}
+
+// Close is used to close connection
+func (d *DirectAdapter) Close() {
+	d.conn.Close()
+}
+
+// Close is used to close connection
+func (d *DirectAdapter) Conn() net.Conn {
+	return d.conn
+}
+
+type Direct struct {
+}
+
+func (d *Direct) Generator(addr *C.Addr) (adapter C.ProxyAdapter, err error) {
+	c, err := net.Dial("tcp", net.JoinHostPort(addr.String(), addr.Port))
+	if err != nil {
+		return
+	}
+	c.(*net.TCPConn).SetKeepAlive(true)
+	return &DirectAdapter{conn: c}, nil
+}
+
+func NewDirect() *Direct {
+	return &Direct{}
+}

adapters/reject.go

@@ -0,0 +1,46 @@
+package adapters
+
+import (
+	"io"
+	"net"
+
+	C "github.com/Dreamacro/clash/constant"
+)
+
+// RejectAdapter is a reject connected adapter
+type RejectAdapter struct {
+}
+
+// ReadWriter is used to handle network traffic
+func (r *RejectAdapter) ReadWriter() io.ReadWriter {
+	return &NopRW{}
+}
+
+// Close is used to close connection
+func (r *RejectAdapter) Close() {}
+
+// Close is used to close connection
+func (r *RejectAdapter) Conn() net.Conn {
+	return nil
+}
+
+type Reject struct {
+}
+
+func (r *Reject) Generator(addr *C.Addr) (adapter C.ProxyAdapter, err error) {
+	return &RejectAdapter{}, nil
+}
+
+func NewReject() *Reject {
+	return &Reject{}
+}
+
+type NopRW struct{}
+
+func (rw *NopRW) Read(b []byte) (int, error) {
+	return len(b), nil
+}
+
+func (rw *NopRW) Write(b []byte) (int, error) {
+	return 0, io.EOF
+}

adapters/shadowsocks.go

@@ -0,0 +1,96 @@
+package adapters
+
+import (
+	"bytes"
+	"fmt"
+	"io"
+	"net"
+	"net/url"
+	"strconv"
+
+	C "github.com/Dreamacro/clash/constant"
+
+	"github.com/riobard/go-shadowsocks2/core"
+	"github.com/riobard/go-shadowsocks2/socks"
+)
+
+// ShadowsocksAdapter is a shadowsocks adapter
+type ShadowsocksAdapter struct {
+	conn net.Conn
+}
+
+// ReadWriter is used to handle network traffic
+func (ss *ShadowsocksAdapter) ReadWriter() io.ReadWriter {
+	return ss.conn
+}
+
+// Close is used to close connection
+func (ss *ShadowsocksAdapter) Close() {
+	ss.conn.Close()
+}
+
+func (ss *ShadowsocksAdapter) Conn() net.Conn {
+	return ss.conn
+}
+
+type ShadowSocks struct {
+	server   string
+	cipher   string
+	password string
+}
+
+func (ss *ShadowSocks) Generator(addr *C.Addr) (adapter C.ProxyAdapter, err error) {
+	var key []byte
+	ciph, _ := core.PickCipher(ss.cipher, key, ss.password)
+	c, err := net.Dial("tcp", ss.server)
+	if err != nil {
+		return nil, fmt.Errorf("%s connect error", ss.server)
+	}
+	c.(*net.TCPConn).SetKeepAlive(true)
+	c = ciph.StreamConn(c)
+	_, err = c.Write(serializesSocksAddr(addr))
+	return &ShadowsocksAdapter{conn: c}, err
+}
+
+func NewShadowSocks(ssURL string) *ShadowSocks {
+	server, cipher, password, _ := parseURL(ssURL)
+	return &ShadowSocks{
+		server:   server,
+		cipher:   cipher,
+		password: password,
+	}
+}
+
+func parseURL(s string) (addr, cipher, password string, err error) {
+	u, err := url.Parse(s)
+	if err != nil {
+		return
+	}
+
+	addr = u.Host
+	if u.User != nil {
+		cipher = u.User.Username()
+		password, _ = u.User.Password()
+	}
+	return
+}
+
+func serializesSocksAddr(addr *C.Addr) []byte {
+	var buf [][]byte
+	aType := uint8(addr.AddrType)
+	p, _ := strconv.Atoi(addr.Port)
+	port := []byte{uint8(p >> 8), uint8(p & 0xff)}
+	switch addr.AddrType {
+	case socks.AtypDomainName:
+		len := uint8(len(addr.Host))
+		host := []byte(addr.Host)
+		buf = [][]byte{[]byte{aType, len}, host, port}
+	case socks.AtypIPv4:
+		host := addr.IP.To4()
+		buf = [][]byte{[]byte{aType}, host, port}
+	case socks.AtypIPv6:
+		host := addr.IP.To16()
+		buf = [][]byte{[]byte{aType}, host, port}
+	}
+	return bytes.Join(buf, []byte(""))
+}

constant/adapters.go

@@ -0,0 +1,22 @@
+package constant
+
+import (
+	"io"
+	"net"
+)
+
+type ProxyAdapter interface {
+	ReadWriter() io.ReadWriter
+	Conn() net.Conn
+	Close()
+}
+
+type ServerAdapter interface {
+	Addr() *Addr
+	Connect(ProxyAdapter)
+	Close()
+}
+
+type Proxy interface {
+	Generator(addr *Addr) (ProxyAdapter, error)
+}

constant/addr.go

@@ -0,0 +1,40 @@
+package constant
+
+import (
+	"net"
+)
+
+// Socks addr type
+const (
+	AtypIPv4       = 1
+	AtypDomainName = 3
+	AtypIPv6       = 4
+
+	TCP = iota
+	UDP
+)
+
+type NetWork int
+
+func (n *NetWork) String() string {
+	if *n == TCP {
+		return "tcp"
+	}
+	return "udp"
+}
+
+// Addr is used to store connection address
+type Addr struct {
+	NetWork  NetWork
+	AddrType int
+	Host     string
+	IP       *net.IP
+	Port     string
+}
+
+func (addr *Addr) String() string {
+	if addr.Host == "" {
+		return addr.IP.String()
+	}
+	return addr.Host
+}

constant/config.go

@@ -0,0 +1,107 @@
+package constant
+
+import (
+	"archive/tar"
+	"compress/gzip"
+	"io"
+	"net/http"
+	"os"
+	"os/user"
+	"path"
+	"strings"
+
+	log "github.com/sirupsen/logrus"
+	"gopkg.in/ini.v1"
+)
+
+const (
+	Name             = "clash"
+	DefalutHTTPPort  = "7890"
+	DefalutSOCKSPort = "7891"
+)
+
+var (
+	HomeDir    string
+	ConfigPath string
+	MMDBPath   string
+)
+
+func init() {
+	currentUser, err := user.Current()
+	if err != nil {
+		log.Fatalf("Can't get current user: %s", err.Error())
+	}
+	HomeDir = currentUser.HomeDir
+
+	dirPath := path.Join(HomeDir, ".config", Name)
+	if _, err := os.Stat(dirPath); os.IsNotExist(err) {
+		if err := os.MkdirAll(dirPath, 0777); err != nil {
+			log.Fatalf("Can't create config directory %s: %s", dirPath, err.Error())
+		}
+	}
+
+	ConfigPath = path.Join(dirPath, "config.ini")
+	if _, err := os.Stat(ConfigPath); os.IsNotExist(err) {
+		log.Info("Can't find config, create a empty file")
+		os.OpenFile(ConfigPath, os.O_CREATE|os.O_WRONLY, 0644)
+	}
+
+	MMDBPath = path.Join(dirPath, "Country.mmdb")
+	if _, err := os.Stat(MMDBPath); os.IsNotExist(err) {
+		log.Info("Can't find MMDB, start download")
+		err := downloadMMDB(MMDBPath)
+		if err != nil {
+			log.Fatalf("Can't download MMDB: %s", err.Error())
+		}
+	}
+}
+
+func downloadMMDB(path string) (err error) {
+	resp, err := http.Get("http://geolite.maxmind.com/download/geoip/database/GeoLite2-Country.tar.gz")
+	if err != nil {
+		return
+	}
+	defer resp.Body.Close()
+
+	gr, err := gzip.NewReader(resp.Body)
+	if err != nil {
+		return
+	}
+	defer gr.Close()
+
+	tr := tar.NewReader(gr)
+	for {
+		h, err := tr.Next()
+		if err == io.EOF {
+			break
+		} else if err != nil {
+			return err
+		}
+
+		if !strings.HasSuffix(h.Name, "GeoLite2-Country.mmdb") {
+			continue
+		}
+
+		f, err := os.OpenFile(path, os.O_CREATE|os.O_WRONLY, 0644)
+		if err != nil {
+			return err
+		}
+		defer f.Close()
+		_, err = io.Copy(f, tr)
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func GetConfig() (*ini.File, error) {
+	if _, err := os.Stat(ConfigPath); os.IsNotExist(err) {
+		return nil, err
+	}
+	return ini.LoadSources(
+		ini.LoadOptions{AllowBooleanKeys: true},
+		ConfigPath,
+	)
+}

constant/rule.go

@@ -0,0 +1,18 @@
+package constant
+
+// Rule Type
+const (
+	DomainSuffix RuleType = iota
+	DomainKeyword
+	GEOIP
+	IPCIDR
+	FINAL
+)
+
+type RuleType int
+
+type Rule interface {
+	RuleType() RuleType
+	IsMatch(addr *Addr) bool
+	Adapter() string
+}

main.go

@@ -0,0 +1,43 @@
+package main
+
+import (
+	"os"
+	"os/signal"
+	"syscall"
+
+	C "github.com/Dreamacro/clash/constant"
+	"github.com/Dreamacro/clash/proxy/http"
+	"github.com/Dreamacro/clash/proxy/socks"
+	"github.com/Dreamacro/clash/tunnel"
+
+	log "github.com/sirupsen/logrus"
+)
+
+func main() {
+	cfg, err := C.GetConfig()
+	if err != nil {
+		log.Fatalf("Read config error: %s", err.Error())
+	}
+
+	port, socksPort := C.DefalutHTTPPort, C.DefalutSOCKSPort
+	section := cfg.Section("General")
+	if key, err := section.GetKey("port"); err == nil {
+		port = key.Value()
+	}
+
+	if key, err := section.GetKey("socks-port"); err == nil {
+		socksPort = key.Value()
+	}
+
+	err = tunnel.GetInstance().UpdateConfig()
+	if err != nil {
+		log.Fatalf("Parse config error: %s", err.Error())
+	}
+
+	go http.NewHttpProxy(port)
+	go socks.NewSocksProxy(socksPort)
+
+	sigCh := make(chan os.Signal, 1)
+	signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
+	<-sigCh
+}

observable/iterable.go

@@ -0,0 +1,18 @@
+package observable
+
+import (
+	"errors"
+)
+
+type Iterable <-chan interface{}
+
+func NewIterable(any interface{}) (Iterable, error) {
+	switch any := any.(type) {
+	case chan interface{}:
+		return Iterable(any), nil
+	case <-chan interface{}:
+		return Iterable(any), nil
+	default:
+		return nil, errors.New("type error")
+	}
+}

observable/observable.go

@@ -0,0 +1,68 @@
+package observable
+
+import (
+	"errors"
+	"sync"
+)
+
+type Observable struct {
+	iterable Iterable
+	listener *sync.Map
+	done     bool
+	doneLock sync.RWMutex
+}
+
+func (o *Observable) process() {
+	for item := range o.iterable {
+		o.listener.Range(func(key, value interface{}) bool {
+			elm := value.(*Subscriber)
+			elm.Emit(item)
+			return true
+		})
+	}
+	o.close()
+}
+
+func (o *Observable) close() {
+	o.doneLock.Lock()
+	o.done = true
+	o.doneLock.Unlock()
+
+	o.listener.Range(func(key, value interface{}) bool {
+		elm := value.(*Subscriber)
+		elm.Close()
+		return true
+	})
+}
+
+func (o *Observable) Subscribe() (Subscription, error) {
+	o.doneLock.RLock()
+	done := o.done
+	o.doneLock.RUnlock()
+	if done == true {
+		return nil, errors.New("Observable is closed")
+	}
+	subscriber := newSubscriber()
+	o.listener.Store(subscriber.Out(), subscriber)
+	return subscriber.Out(), nil
+}
+
+func (o *Observable) UnSubscribe(sub Subscription) {
+	elm, exist := o.listener.Load(sub)
+	if !exist {
+		println("not exist")
+		return
+	}
+	subscriber := elm.(*Subscriber)
+	o.listener.Delete(subscriber.Out())
+	subscriber.Close()
+}
+
+func NewObservable(any Iterable) *Observable {
+	observable := &Observable{
+		iterable: any,
+		listener: &sync.Map{},
+	}
+	go observable.process()
+	return observable
+}

observable/observable_test.go

@@ -0,0 +1,117 @@
+package observable
+
+import (
+	"runtime"
+	"sync"
+	"testing"
+	"time"
+)
+
+func iterator(item []interface{}) chan interface{} {
+	ch := make(chan interface{})
+	go func() {
+		time.Sleep(100 * time.Millisecond)
+		for _, elm := range item {
+			ch <- elm
+		}
+		close(ch)
+	}()
+	return ch
+}
+
+func TestObservable(t *testing.T) {
+	iter := iterator([]interface{}{1, 2, 3, 4, 5})
+	src := NewObservable(iter)
+	data, err := src.Subscribe()
+	if err != nil {
+		t.Error(err)
+	}
+	count := 0
+	for {
+		_, open := <-data
+		if !open {
+			break
+		}
+		count = count + 1
+	}
+	if count != 5 {
+		t.Error("Revc number error")
+	}
+}
+
+func TestObservable_MutilSubscribe(t *testing.T) {
+	iter := iterator([]interface{}{1, 2, 3, 4, 5})
+	src := NewObservable(iter)
+	ch1, _ := src.Subscribe()
+	ch2, _ := src.Subscribe()
+	count := 0
+
+	var wg sync.WaitGroup
+	wg.Add(2)
+	waitCh := func(ch <-chan interface{}) {
+		for {
+			_, open := <-ch
+			if !open {
+				break
+			}
+			count = count + 1
+		}
+		wg.Done()
+	}
+	go waitCh(ch1)
+	go waitCh(ch2)
+	wg.Wait()
+	if count != 10 {
+		t.Error("Revc number error")
+	}
+}
+
+func TestObservable_UnSubscribe(t *testing.T) {
+	iter := iterator([]interface{}{1, 2, 3, 4, 5})
+	src := NewObservable(iter)
+	data, err := src.Subscribe()
+	if err != nil {
+		t.Error(err)
+	}
+	src.UnSubscribe(data)
+	_, open := <-data
+	if open {
+		t.Error("Revc number error")
+	}
+}
+
+func TestObservable_SubscribeGoroutineLeak(t *testing.T) {
+	// waiting for other goroutine recycle
+	time.Sleep(120 * time.Millisecond)
+	init := runtime.NumGoroutine()
+	iter := iterator([]interface{}{1, 2, 3, 4, 5})
+	src := NewObservable(iter)
+	max := 100
+
+	var list []Subscription
+	for i := 0; i < max; i++ {
+		ch, _ := src.Subscribe()
+		list = append(list, ch)
+	}
+
+	var wg sync.WaitGroup
+	wg.Add(max)
+	waitCh := func(ch <-chan interface{}) {
+		for {
+			_, open := <-ch
+			if !open {
+				break
+			}
+		}
+		wg.Done()
+	}
+
+	for _, ch := range list {
+		go waitCh(ch)
+	}
+	wg.Wait()
+	now := runtime.NumGoroutine()
+	if init != now {
+		t.Errorf("Goroutine Leak: init %d now %d", init, now)
+	}
+}

observable/subscriber.go

@@ -0,0 +1,35 @@
+package observable
+
+import (
+	"sync"
+
+	"gopkg.in/eapache/channels.v1"
+)
+
+type Subscription <-chan interface{}
+
+type Subscriber struct {
+	buffer *channels.InfiniteChannel
+	once   sync.Once
+}
+
+func (s *Subscriber) Emit(item interface{}) {
+	s.buffer.In() <- item
+}
+
+func (s *Subscriber) Out() Subscription {
+	return s.buffer.Out()
+}
+
+func (s *Subscriber) Close() {
+	s.once.Do(func() {
+		s.buffer.Close()
+	})
+}
+
+func newSubscriber() *Subscriber {
+	sub := &Subscriber{
+		buffer: channels.NewInfiniteChannel(),
+	}
+	return sub
+}

observable/util.go

@@ -0,0 +1,15 @@
+package observable
+
+func mergeWithBytes(ch <-chan interface{}, buf []byte) chan interface{} {
+	out := make(chan interface{})
+	go func() {
+		defer close(out)
+		if len(buf) != 0 {
+			out <- buf
+		}
+		for elm := range ch {
+			out <- elm
+		}
+	}()
+	return out
+}

proxy/http/http.go

@@ -0,0 +1,62 @@
+package http
+
+import (
+	"io"
+	"net"
+	"net/http"
+	"time"
+
+	C "github.com/Dreamacro/clash/constant"
+)
+
+type HttpAdapter struct {
+	addr *C.Addr
+	r    *http.Request
+	w    http.ResponseWriter
+	done chan struct{}
+}
+
+func (h *HttpAdapter) Close() {
+	h.done <- struct{}{}
+}
+
+func (h *HttpAdapter) Addr() *C.Addr {
+	return h.addr
+}
+
+func (h *HttpAdapter) Connect(proxy C.ProxyAdapter) {
+	req := http.Transport{
+		Dial: func(string, string) (net.Conn, error) {
+			return proxy.Conn(), nil
+		},
+		// from http.DefaultTransport
+		MaxIdleConns:          100,
+		IdleConnTimeout:       90 * time.Second,
+		TLSHandshakeTimeout:   10 * time.Second,
+		ExpectContinueTimeout: 1 * time.Second,
+	}
+	resp, err := req.RoundTrip(h.r)
+	if err != nil {
+		return
+	}
+	defer resp.Body.Close()
+
+	header := h.w.Header()
+	for k, vv := range resp.Header {
+		for _, v := range vv {
+			header.Add(k, v)
+		}
+	}
+	h.w.WriteHeader(resp.StatusCode)
+	io.Copy(h.w, resp.Body)
+}
+
+func NewHttp(host string, w http.ResponseWriter, r *http.Request) (*HttpAdapter, chan struct{}) {
+	done := make(chan struct{})
+	return &HttpAdapter{
+		addr: parseHttpAddr(host),
+		r:    r,
+		w:    w,
+		done: done,
+	}, done
+}

proxy/http/https.go

@@ -0,0 +1,35 @@
+package http
+
+import (
+	"bufio"
+	"io"
+	"net"
+
+	C "github.com/Dreamacro/clash/constant"
+)
+
+type HttpsAdapter struct {
+	addr *C.Addr
+	conn net.Conn
+	rw   *bufio.ReadWriter
+}
+
+func (h *HttpsAdapter) Close() {
+	h.conn.Close()
+}
+
+func (h *HttpsAdapter) Addr() *C.Addr {
+	return h.addr
+}
+
+func (h *HttpsAdapter) Connect(proxy C.ProxyAdapter) {
+	go io.Copy(h.conn, proxy.ReadWriter())
+	io.Copy(proxy.ReadWriter(), h.conn)
+}
+
+func NewHttps(host string, conn net.Conn) *HttpsAdapter {
+	return &HttpsAdapter{
+		addr: parseHttpAddr(host),
+		conn: conn,
+	}
+}

proxy/http/server.go

@@ -0,0 +1,86 @@
+package http
+
+import (
+	"fmt"
+	"net"
+	"net/http"
+	"strings"
+
+	C "github.com/Dreamacro/clash/constant"
+	"github.com/Dreamacro/clash/tunnel"
+
+	"github.com/riobard/go-shadowsocks2/socks"
+	log "github.com/sirupsen/logrus"
+)
+
+var (
+	tun = tunnel.GetInstance()
+)
+
+func NewHttpProxy(port string) {
+	server := &http.Server{
+		Addr: fmt.Sprintf(":%s", port),
+		Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			if r.Method == http.MethodConnect {
+				handleTunneling(w, r)
+			} else {
+				handleHTTP(w, r)
+			}
+		}),
+	}
+	log.Infof("HTTP proxy :%s", port)
+	server.ListenAndServe()
+}
+
+func handleHTTP(w http.ResponseWriter, r *http.Request) {
+	addr := r.Host
+	// padding default port
+	if !strings.Contains(addr, ":") {
+		addr += ":80"
+	}
+	req, done := NewHttp(addr, w, r)
+	tun.Add(req)
+	<-done
+}
+
+func handleTunneling(w http.ResponseWriter, r *http.Request) {
+	hijacker, ok := w.(http.Hijacker)
+	if !ok {
+		return
+	}
+	conn, _, err := hijacker.Hijack()
+	if err != nil {
+		return
+	}
+	// w.WriteHeader(http.StatusOK) doesn't works in Safari
+	conn.Write([]byte("HTTP/1.1 200 OK\r\n\r\n"))
+	tun.Add(NewHttps(r.Host, conn))
+}
+
+func parseHttpAddr(target string) *C.Addr {
+	host, port, _ := net.SplitHostPort(target)
+	ipAddr, err := net.ResolveIPAddr("ip", host)
+	var resolveIP *net.IP
+	if err == nil {
+		resolveIP = &ipAddr.IP
+	}
+
+	var addType int
+	ip := net.ParseIP(host)
+	switch {
+	case ip == nil:
+		addType = socks.AtypDomainName
+	case ip.To4() == nil:
+		addType = socks.AtypIPv6
+	default:
+		addType = socks.AtypIPv4
+	}
+
+	return &C.Addr{
+		NetWork:  C.TCP,
+		AddrType: addType,
+		Host:     host,
+		IP:       resolveIP,
+		Port:     port,
+	}
+}

proxy/socks/tcp.go

@@ -0,0 +1,98 @@
+package socks
+
+import (
+	"fmt"
+	"io"
+	"net"
+	"strconv"
+
+	C "github.com/Dreamacro/clash/constant"
+	"github.com/Dreamacro/clash/tunnel"
+
+	"github.com/riobard/go-shadowsocks2/socks"
+	log "github.com/sirupsen/logrus"
+)
+
+var (
+	tun = tunnel.GetInstance()
+)
+
+func NewSocksProxy(port string) {
+	l, err := net.Listen("tcp", fmt.Sprintf(":%s", port))
+	defer l.Close()
+	if err != nil {
+		return
+	}
+	log.Infof("SOCKS proxy :%s", port)
+	for {
+		c, err := l.Accept()
+		if err != nil {
+			continue
+		}
+		go handleSocks(c)
+	}
+}
+
+func handleSocks(conn net.Conn) {
+	target, err := socks.Handshake(conn)
+	if err != nil {
+		conn.Close()
+		return
+	}
+	conn.(*net.TCPConn).SetKeepAlive(true)
+	tun.Add(NewSocks(target, conn))
+}
+
+type SocksAdapter struct {
+	conn net.Conn
+	addr *C.Addr
+}
+
+func (s *SocksAdapter) Close() {
+	s.conn.Close()
+}
+
+func (s *SocksAdapter) Addr() *C.Addr {
+	return s.addr
+}
+
+func (s *SocksAdapter) Connect(proxy C.ProxyAdapter) {
+	go io.Copy(s.conn, proxy.ReadWriter())
+	io.Copy(proxy.ReadWriter(), s.conn)
+}
+
+func parseSocksAddr(target socks.Addr) *C.Addr {
+	var host, port string
+	var ip net.IP
+
+	switch target[0] {
+	case socks.AtypDomainName:
+		host = string(target[2 : 2+target[1]])
+		port = strconv.Itoa((int(target[2+target[1]]) << 8) | int(target[2+target[1]+1]))
+		ipAddr, err := net.ResolveIPAddr("ip", host)
+		if err == nil {
+			ip = ipAddr.IP
+		}
+	case socks.AtypIPv4:
+		ip = net.IP(target[1 : 1+net.IPv4len])
+		port = strconv.Itoa((int(target[1+net.IPv4len]) << 8) | int(target[1+net.IPv4len+1]))
+	case socks.AtypIPv6:
+		ip = net.IP(target[1 : 1+net.IPv6len])
+		port = strconv.Itoa((int(target[1+net.IPv6len]) << 8) | int(target[1+net.IPv6len+1]))
+	}
+
+	return &C.Addr{
+		NetWork:  C.TCP,
+		AddrType: int(target[0]),
+		Host:     host,
+		IP:       &ip,
+		Port:     port,
+	}
+}
+
+func NewSocks(target socks.Addr, conn net.Conn) *SocksAdapter {
+	return &SocksAdapter{
+		conn: conn,
+		addr: parseSocksAddr(target),
+	}
+}

proxy/socks/udp.go

@@ -0,0 +1 @@
+package socks

rules/domain_keyword.go

@@ -0,0 +1,35 @@
+package rules
+
+import (
+	"strings"
+
+	C "github.com/Dreamacro/clash/constant"
+)
+
+type DomainKeyword struct {
+	keyword string
+	adapter string
+}
+
+func (dk *DomainKeyword) RuleType() C.RuleType {
+	return C.DomainKeyword
+}
+
+func (dk *DomainKeyword) IsMatch(addr *C.Addr) bool {
+	if addr.AddrType != C.AtypDomainName {
+		return false
+	}
+	domain := addr.Host
+	return strings.Contains(domain, dk.keyword)
+}
+
+func (dk *DomainKeyword) Adapter() string {
+	return dk.adapter
+}
+
+func NewDomainKeyword(keyword string, adapter string) *DomainKeyword {
+	return &DomainKeyword{
+		keyword: keyword,
+		adapter: adapter,
+	}
+}

rules/domain_suffix.go

@@ -0,0 +1,35 @@
+package rules
+
+import (
+	"strings"
+
+	C "github.com/Dreamacro/clash/constant"
+)
+
+type DomainSuffix struct {
+	suffix  string
+	adapter string
+}
+
+func (ds *DomainSuffix) RuleType() C.RuleType {
+	return C.DomainSuffix
+}
+
+func (ds *DomainSuffix) IsMatch(addr *C.Addr) bool {
+	if addr.AddrType != C.AtypDomainName {
+		return false
+	}
+	domain := addr.Host
+	return strings.HasSuffix(domain, "."+ds.suffix) || domain == ds.suffix
+}
+
+func (ds *DomainSuffix) Adapter() string {
+	return ds.adapter
+}
+
+func NewDomainSuffix(suffix string, adapter string) *DomainSuffix {
+	return &DomainSuffix{
+		suffix:  suffix,
+		adapter: adapter,
+	}
+}

rules/final.go

@@ -0,0 +1,27 @@
+package rules
+
+import (
+	C "github.com/Dreamacro/clash/constant"
+)
+
+type Final struct {
+	adapter string
+}
+
+func (f *Final) RuleType() C.RuleType {
+	return C.FINAL
+}
+
+func (f *Final) IsMatch(addr *C.Addr) bool {
+	return true
+}
+
+func (f *Final) Adapter() string {
+	return f.adapter
+}
+
+func NewFinal(adapter string) *Final {
+	return &Final{
+		adapter: adapter,
+	}
+}

rules/geoip.go

@@ -0,0 +1,46 @@
+package rules
+
+import (
+	C "github.com/Dreamacro/clash/constant"
+
+	"github.com/oschwald/geoip2-golang"
+	log "github.com/sirupsen/logrus"
+)
+
+var mmdb *geoip2.Reader
+
+func init() {
+	var err error
+	mmdb, err = geoip2.Open(C.MMDBPath)
+	if err != nil {
+		log.Fatalf("Can't load mmdb: %s", err.Error())
+	}
+}
+
+type GEOIP struct {
+	country string
+	adapter string
+}
+
+func (g *GEOIP) RuleType() C.RuleType {
+	return C.GEOIP
+}
+
+func (g *GEOIP) IsMatch(addr *C.Addr) bool {
+	if addr.IP == nil {
+		return false
+	}
+	record, _ := mmdb.Country(*addr.IP)
+	return record.Country.IsoCode == g.country
+}
+
+func (g *GEOIP) Adapter() string {
+	return g.adapter
+}
+
+func NewGEOIP(country string, adapter string) *GEOIP {
+	return &GEOIP{
+		country: country,
+		adapter: adapter,
+	}
+}

rules/ipcidr.go

@@ -0,0 +1,38 @@
+package rules
+
+import (
+	"net"
+
+	C "github.com/Dreamacro/clash/constant"
+)
+
+type IPCIDR struct {
+	ipnet   *net.IPNet
+	adapter string
+}
+
+func (i *IPCIDR) RuleType() C.RuleType {
+	return C.IPCIDR
+}
+
+func (i *IPCIDR) IsMatch(addr *C.Addr) bool {
+	if addr.IP == nil {
+		return false
+	}
+
+	return i.ipnet.Contains(*addr.IP)
+}
+
+func (g *IPCIDR) Adapter() string {
+	return g.adapter
+}
+
+func NewIPCIDR(s string, adapter string) *IPCIDR {
+	_, ipnet, err := net.ParseCIDR(s)
+	if err != nil {
+	}
+	return &IPCIDR{
+		ipnet:   ipnet,
+		adapter: adapter,
+	}
+}

tunnel/log.go

@@ -0,0 +1,52 @@
+package tunnel
+
+import (
+	"fmt"
+
+	log "github.com/sirupsen/logrus"
+)
+
+const (
+	INFO LogType = iota
+	WARNING
+	ERROR
+	DEBUG
+)
+
+type LogType int
+
+type Log struct {
+	LogType LogType
+	Payload string
+}
+
+func print(data Log) {
+	switch data.LogType {
+	case INFO:
+		log.Infoln(data.Payload)
+	case WARNING:
+		log.Warnln(data.Payload)
+	case ERROR:
+		log.Errorln(data.Payload)
+	case DEBUG:
+		log.Debugln(data.Payload)
+	}
+}
+
+func (t *Tunnel) subscribeLogs() {
+	sub, err := t.observable.Subscribe()
+	if err != nil {
+		log.Fatalf("Can't subscribe tunnel log: %s", err.Error())
+	}
+	for elm := range sub {
+		data := elm.(Log)
+		print(data)
+	}
+}
+
+func newLog(logType LogType, format string, v ...interface{}) Log {
+	return Log{
+		LogType: logType,
+		Payload: fmt.Sprintf(format, v...),
+	}
+}

tunnel/tunnel.go

@@ -0,0 +1,156 @@
+package tunnel
+
+import (
+	"fmt"
+	"strings"
+	"sync"
+
+	"github.com/Dreamacro/clash/adapters"
+	C "github.com/Dreamacro/clash/constant"
+	"github.com/Dreamacro/clash/observable"
+	R "github.com/Dreamacro/clash/rules"
+
+	"gopkg.in/eapache/channels.v1"
+)
+
+var (
+	tunnel *Tunnel
+	once   sync.Once
+)
+
+type Tunnel struct {
+	queue      *channels.InfiniteChannel
+	rules      []C.Rule
+	proxys     map[string]C.Proxy
+	observable *observable.Observable
+	logCh      chan interface{}
+	configLock *sync.RWMutex
+}
+
+func (t *Tunnel) Add(req C.ServerAdapter) {
+	t.queue.In() <- req
+}
+
+func (t *Tunnel) UpdateConfig() (err error) {
+	t.configLock.Lock()
+	defer t.configLock.Unlock()
+
+	cfg, err := C.GetConfig()
+	if err != nil {
+		return
+	}
+
+	// clear proxys and rules
+	t.proxys = make(map[string]C.Proxy)
+	t.rules = []C.Rule{}
+
+	proxys := cfg.Section("Proxy")
+	rules := cfg.Section("Rule")
+
+	// parse proxy
+	for _, key := range proxys.Keys() {
+		proxy := strings.Split(key.Value(), ",")
+		if len(proxy) == 0 {
+			continue
+		}
+		proxy = trimArr(proxy)
+		switch proxy[0] {
+		// ss, server, port, cipter, password
+		case "ss":
+			if len(proxy) < 5 {
+				continue
+			}
+			ssURL := fmt.Sprintf("ss://%s:%s@%s:%s", proxy[3], proxy[4], proxy[1], proxy[2])
+			t.proxys[key.Name()] = adapters.NewShadowSocks(ssURL)
+		}
+	}
+
+	// init proxy
+	t.proxys["DIRECT"] = adapters.NewDirect()
+	t.proxys["REJECT"] = adapters.NewReject()
+
+	// parse rules
+	for _, key := range rules.Keys() {
+		rule := strings.Split(key.Name(), ",")
+		if len(rule) < 3 {
+			continue
+		}
+		rule = trimArr(rule)
+		switch rule[0] {
+		case "DOMAIN-SUFFIX":
+			t.rules = append(t.rules, R.NewDomainSuffix(rule[1], rule[2]))
+		case "DOMAIN-KEYWORD":
+			t.rules = append(t.rules, R.NewDomainKeyword(rule[1], rule[2]))
+		case "GEOIP":
+			t.rules = append(t.rules, R.NewGEOIP(rule[1], rule[2]))
+		case "IP-CIDR", "IP-CIDR6":
+			t.rules = append(t.rules, R.NewIPCIDR(rule[1], rule[2]))
+		case "FINAL":
+			t.rules = append(t.rules, R.NewFinal(rule[2]))
+		}
+	}
+
+	return nil
+}
+
+func (t *Tunnel) process() {
+	queue := t.queue.Out()
+	for {
+		elm := <-queue
+		conn := elm.(C.ServerAdapter)
+		go t.handleConn(conn)
+	}
+}
+
+func (t *Tunnel) handleConn(localConn C.ServerAdapter) {
+	defer localConn.Close()
+	addr := localConn.Addr()
+	proxy := t.match(addr)
+	remoConn, err := proxy.Generator(addr)
+	if err != nil {
+		t.logCh <- newLog(WARNING, "Proxy connect error: %s", err.Error())
+		return
+	}
+	defer remoConn.Close()
+
+	localConn.Connect(remoConn)
+}
+
+func (t *Tunnel) match(addr *C.Addr) C.Proxy {
+	t.configLock.RLock()
+	defer t.configLock.RUnlock()
+
+	for _, rule := range t.rules {
+		if rule.IsMatch(addr) {
+			a, ok := t.proxys[rule.Adapter()]
+			if !ok {
+				continue
+			}
+			t.logCh <- newLog(INFO, "%v match %d using %s", addr.String(), rule.RuleType(), rule.Adapter())
+			return a
+		}
+	}
+	t.logCh <- newLog(INFO, "don't find, direct")
+	return t.proxys["DIRECT"]
+}
+
+func newTunnel() *Tunnel {
+	logCh := make(chan interface{})
+	tunnel := &Tunnel{
+		queue:      channels.NewInfiniteChannel(),
+		proxys:     make(map[string]C.Proxy),
+		observable: observable.NewObservable(logCh),
+		logCh:      logCh,
+		configLock: &sync.RWMutex{},
+	}
+	go tunnel.process()
+	go tunnel.subscribeLogs()
+	return tunnel
+}
+
+func GetInstance() *Tunnel {
+	once.Do(func() {
+		tunnel = newTunnel()
+	})
+	return tunnel
+}

tunnel/utils.go

@@ -0,0 +1,12 @@
+package tunnel
+
+import (
+	"strings"
+)
+
+func trimArr(arr []string) (r []string) {
+	for _, e := range arr {
+		r = append(r, strings.Trim(e, " "))
+	}
+	return
+}