将版本修改为0.3.1

build/pages/403.html

@@ -9,7 +9,7 @@
 <h3>403 Forbidden</h3>
 <p>Sorry, your access to the page has been denied. Please try again later.</p>
 
-<footer>Powered by TeaEdge.</footer>
+<footer>Powered by GoEdge.</footer>
 
 </body>
 </html>

build/pages/404.html

@@ -9,7 +9,7 @@
 <h3>404 Not Found</h3>
 <p>Sorry, the page you are looking for is not found. Please try again later.</p>
 
-<footer>Powered by TeaEdge.</footer>
+<footer>Powered by GoEdge.</footer>
 
 </body>
 </html>

build/pages/50x.html

@@ -9,7 +9,7 @@
 <h3>An error occurred.</h3>
 <p>Sorry, the page you are looking for is currently unavailable. Please try again later.</p>
 
-<footer>Powered by TeaEdge.</footer>
+<footer>Powered by GoEdge.</footer>
 
 </body>
 </html>

build/pages/shutdown_en.html

@@ -9,7 +9,7 @@
 <h3>The website is shutdown.</h3>
 <p>Sorry, the page you are looking for is currently unavailable. Please try again later.</p>
 
-<footer>Powered by TeaEdge.</footer>
+<footer>Powered by GoEdge.</footer>
 
 </body>
 </html>

build/pages/shutdown_upgrade_zh.html

@@ -9,7 +9,7 @@
 <h3>网站升级中</h3>
 <p>为了给您提供更好的服务，我们正在升级网站，请稍后重新访问。</p>
 
-<footer>Powered by TeaEdge.</footer>
+<footer>Powered by GoEdge.</footer>
 
 </body>
 </html>

build/pages/shutdown_zh.html

@@ -9,7 +9,7 @@
 <h3>网站暂时关闭</h3>
 <p>网站已被暂时关闭，请耐心等待我们的重新开通通知。</p>
 
-<footer>Powered by TeaEdge.</footer>
+<footer>Powered by GoEdge.</footer>
 
 </body>
 </html>

internal/caches/reader_file.go

@@ -63,6 +63,7 @@ func (this *FileReader) Init() error {
 	// body
 	bodySize := int(binary.BigEndian.Uint64(buf[SizeExpiresAt+SizeStatus+SizeURLLength+SizeHeaderLength : SizeExpiresAt+SizeStatus+SizeURLLength+SizeHeaderLength+SizeBodyLength]))
 	if bodySize == 0 {
+		isOk = true
 		return nil
 	}
 	this.bodySize = int64(bodySize)

internal/const/const.go

@@ -1,7 +1,7 @@
 package teaconst
 
 const (
-	Version = "0.3.0"
+	Version = "0.3.1"
 
 	ProductName = "Edge Node"
 	ProcessName = "edge-node"

internal/iplibrary/manager_country.go

@@ -20,12 +20,12 @@ import (
 var SharedCountryManager = NewCountryManager()
 
 func init() {
-	events.On(events.EventStart, func() {
+	events.On(events.EventLoaded, func() {
 		go SharedCountryManager.Start()
 	})
 }
 
-// 国家信息管理
+// CountryManager 国家/地区信息管理
 type CountryManager struct {
 	cacheFile string
 

internal/iplibrary/manager_ip_list.go

@@ -15,12 +15,12 @@ var SharedIPListManager = NewIPListManager()
 var IPListUpdateNotify = make(chan bool, 1)
 
 func init() {
-	events.On(events.EventStart, func() {
+	events.On(events.EventLoaded, func() {
 		go SharedIPListManager.Start()
 	})
 }
 
-// IP名单管理
+// IPListManager IP名单管理
 type IPListManager struct {
 	// 缓存文件
 	// 每行一个数据：id|from|to|expiredAt

internal/iplibrary/manager_province.go

@@ -24,12 +24,12 @@ const (
 var SharedProvinceManager = NewProvinceManager()
 
 func init() {
-	events.On(events.EventStart, func() {
+	events.On(events.EventLoaded, func() {
 		go SharedProvinceManager.Start()
 	})
 }
 
-// 国家信息管理
+// ProvinceManager 中国省份信息管理
 type ProvinceManager struct {
 	cacheFile string
 

internal/monitor/value_queue.go

@@ -15,7 +15,7 @@ import (
 var SharedValueQueue = NewValueQueue()
 
 func init() {
-	events.On(events.EventStart, func() {
+	events.On(events.EventLoaded, func() {
 		go SharedValueQueue.Start()
 	})
 }

internal/nodes/http_request_page.go

@@ -2,9 +2,9 @@ package nodes
 
 import (
 	"github.com/TeaOSLab/EdgeNode/internal/remotelogs"
+	"github.com/TeaOSLab/EdgeNode/internal/utils"
 	"github.com/iwind/TeaGo/Tea"
 	"github.com/iwind/TeaGo/logs"
-	"io"
 	"net/http"
 	"os"
 	"regexp"
@@ -21,7 +21,7 @@ func (this *HTTPRequest) doPage(status int) (shouldStop bool) {
 	for _, page := range this.web.Pages {
 		if page.Match(status) {
 			if urlPrefixRegexp.MatchString(page.URL) {
-				this.doURL(http.MethodGet, page.URL, "", page.NewStatus)
+				this.doURL(http.MethodGet, page.URL, "", page.NewStatus, true)
 				return true
 			} else {
 				file := Tea.Root + Tea.DS + page.URL
@@ -48,7 +48,9 @@ func (this *HTTPRequest) doPage(status int) (shouldStop bool) {
 					this.writer.WriteHeader(status)
 				}
 				buf := bytePool1k.Get()
-				_, err = io.CopyBuffer(this.writer, fp, buf)
+				_, err = utils.CopyWithFilter(this.writer, fp, buf, func(p []byte) []byte {
+					return []byte(this.Format(string(p)))
+				})
 				bytePool1k.Put(buf)
 				if err != nil {
 					if !this.canIgnore(err) {

internal/nodes/http_request_reverse_proxy.go

@@ -33,7 +33,11 @@ func (this *HTTPRequest) doReverseProxy() {
 
 	// 源站
 	requestCall := shared.NewRequestCall()
+	requestCall.Request = this.RawReq
+	requestCall.Formatter = this.Format
+	requestCall.Domain = this.Host
 	origin := this.reverseProxy.NextOrigin(requestCall)
+	requestCall.CallResponseCallbacks(this.writer)
 	if origin == nil {
 		err := errors.New(this.requestPath() + ": no available backends for reverse proxy")
 		remotelogs.Error("HTTP_REQUEST_REVERSE_PROXY", err.Error())

internal/nodes/http_request_rewrite.go

@@ -19,7 +19,7 @@ func (this *HTTPRequest) doRewrite() (shouldShop bool) {
 			if len(this.rewriteRule.ProxyHost) > 0 {
 				host = this.rewriteRule.ProxyHost
 			}
-			this.doURL(this.RawReq.Method, this.rewriteReplace, host, 0)
+			this.doURL(this.RawReq.Method, this.rewriteReplace, host, 0, false)
 			return true
 		}
 

internal/nodes/http_request_shutdown.go

@@ -2,9 +2,9 @@ package nodes
 
 import (
 	"github.com/TeaOSLab/EdgeNode/internal/remotelogs"
+	"github.com/TeaOSLab/EdgeNode/internal/utils"
 	"github.com/iwind/TeaGo/Tea"
 	"github.com/iwind/TeaGo/logs"
-	"io"
 	"net/http"
 	"os"
 )
@@ -17,7 +17,7 @@ func (this *HTTPRequest) doShutdown() {
 	}
 
 	if urlPrefixRegexp.MatchString(shutdown.URL) { // URL
-		this.doURL(http.MethodGet, shutdown.URL, "", shutdown.Status)
+		this.doURL(http.MethodGet, shutdown.URL, "", shutdown.Status, true)
 		return
 	}
 
@@ -64,7 +64,9 @@ func (this *HTTPRequest) doShutdown() {
 		this.writer.WriteHeader(http.StatusOK)
 	}
 	buf := bytePool1k.Get()
-	_, err = io.CopyBuffer(this.writer, fp, buf)
+	_, err = utils.CopyWithFilter(this.writer, fp, buf, func(p []byte) []byte {
+		return []byte(this.Format(string(p)))
+	})
 	bytePool1k.Put(buf)
 	if err != nil {
 		if !this.canIgnore(err) {

internal/nodes/http_request_url.go

@@ -1,7 +1,6 @@
 package nodes
 
 import (
-	"errors"
 	"github.com/TeaOSLab/EdgeNode/internal/remotelogs"
 	"github.com/TeaOSLab/EdgeNode/internal/utils"
 	"github.com/iwind/TeaGo/logs"
@@ -11,7 +10,7 @@ import (
 )
 
 // 请求某个URL
-func (this *HTTPRequest) doURL(method string, url string, host string, statusCode int) {
+func (this *HTTPRequest) doURL(method string, url string, host string, statusCode int, supportVariables bool) {
 	req, err := http.NewRequest(method, url, this.RawReq.Body)
 	if err != nil {
 		logs.Error(err)
@@ -35,7 +34,7 @@ func (this *HTTPRequest) doURL(method string, url string, host string, statusCod
 	var client = utils.SharedHttpClient(60 * time.Second)
 	resp, err := client.Do(req)
 	if err != nil {
-		logs.Error(errors.New(req.URL.String() + ": " + err.Error()))
+		remotelogs.Error("HTTP_REQUEST_URL", req.URL.String()+": "+err.Error())
 		this.write50x(err, http.StatusInternalServerError)
 		return
 	}
@@ -50,6 +49,9 @@ func (this *HTTPRequest) doURL(method string, url string, host string, statusCod
 		this.processResponseHeaders(statusCode)
 	}
 
+	if supportVariables {
+		resp.Header.Del("Content-Length")
+	}
 	this.writer.AddHeaders(resp.Header)
 	if statusCode <= 0 {
 		this.writer.Prepare(resp.ContentLength, resp.StatusCode)
@@ -67,7 +69,13 @@ func (this *HTTPRequest) doURL(method string, url string, host string, statusCod
 	// 输出内容
 	pool := this.bytePool(resp.ContentLength)
 	buf := pool.Get()
-	_, err = io.CopyBuffer(this.writer, resp.Body, buf)
+	if supportVariables {
+		_, err = utils.CopyWithFilter(this.writer, resp.Body, buf, func(p []byte) []byte {
+			return []byte(this.Format(string(p)))
+		})
+	} else {
+		_, err = io.CopyBuffer(this.writer, resp.Body, buf)
+	}
 	pool.Put(buf)
 
 	if err != nil {

internal/nodes/listener.go

@@ -11,7 +11,7 @@ import (
 )
 
 type Listener struct {
-	group       *serverconfigs.ServerGroup
+	group       *serverconfigs.ServerAddressGroup
 	isListening bool
 	listener    ListenerInterface // 监听器
 
@@ -22,7 +22,7 @@ func NewListener() *Listener {
 	return &Listener{}
 }
 
-func (this *Listener) Reload(group *serverconfigs.ServerGroup) {
+func (this *Listener) Reload(group *serverconfigs.ServerAddressGroup) {
 	this.locker.Lock()
 	this.group = group
 	if this.listener != nil {

internal/nodes/listener_base.go

@@ -16,7 +16,7 @@ type BaseListener struct {
 	namedServersLocker sync.RWMutex
 	namedServers       map[string]*NamedServer // 域名 => server
 
-	Group *serverconfigs.ServerGroup
+	Group *serverconfigs.ServerAddressGroup
 
 	countActiveConnections int64 // 当前活跃的连接数
 }

internal/nodes/listener_http.go

@@ -102,7 +102,7 @@ func (this *HTTPListener) Close() error {
 	return this.Listener.Close()
 }
 
-func (this *HTTPListener) Reload(group *serverconfigs.ServerGroup) {
+func (this *HTTPListener) Reload(group *serverconfigs.ServerAddressGroup) {
 	this.Group = group
 
 	this.Reset()

internal/nodes/listener_interface.go

@@ -14,7 +14,7 @@ type ListenerInterface interface {
 	Close() error
 
 	// Reload 重载配置
-	Reload(serverGroup *serverconfigs.ServerGroup)
+	Reload(serverGroup *serverconfigs.ServerAddressGroup)
 
 	// CountActiveListeners 获取当前活跃的连接数
 	CountActiveListeners() int

internal/nodes/listener_manager.go

@@ -70,7 +70,7 @@ func (this *ListenerManager) Start(node *nodeconfigs.NodeConfig) error {
 	groupAddrs := []string{}
 	availableServerGroups := node.AvailableGroups()
 	if !node.IsOn {
-		availableServerGroups = []*serverconfigs.ServerGroup{}
+		availableServerGroups = []*serverconfigs.ServerAddressGroup{}
 	}
 
 	if len(availableServerGroups) == 0 {

internal/nodes/listener_tcp.go

@@ -42,7 +42,7 @@ func (this *TCPListener) Serve() error {
 	return nil
 }
 
-func (this *TCPListener) Reload(group *serverconfigs.ServerGroup) {
+func (this *TCPListener) Reload(group *serverconfigs.ServerAddressGroup) {
 	this.Group = group
 	this.Reset()
 }

internal/nodes/listener_udp.go

@@ -84,7 +84,7 @@ func (this *UDPListener) Close() error {
 	return this.Listener.Close()
 }
 
-func (this *UDPListener) Reload(group *serverconfigs.ServerGroup) {
+func (this *UDPListener) Reload(group *serverconfigs.ServerAddressGroup) {
 	this.Group = group
 	this.Reset()
 }

internal/nodes/listener_unix.go

@@ -22,7 +22,7 @@ func (this *UnixListener) Close() error {
 	return nil
 }
 
-func (this *UnixListener) Reload(group *serverconfigs.ServerGroup) {
+func (this *UnixListener) Reload(group *serverconfigs.ServerAddressGroup) {
 	this.Group = group
 	this.Reset()
 }

internal/nodes/node.go

@@ -290,7 +290,7 @@ func (this *Node) syncConfig() error {
 				if os.IsNotExist(clusterErr) {
 					return err
 				}
-				return clusterErr
+				return errors.New("check cluster config failed: " + clusterErr.Error())
 			}
 		} else {
 			return err
@@ -347,7 +347,7 @@ func (this *Node) syncConfig() error {
 	} else {
 		remotelogs.Println("NODE", "loading config ...")
 	}
-	
+
 	nodeconfigs.ResetNodeConfig(nodeConfig)
 	caches.SharedManager.MaxDiskCapacity = nodeConfig.MaxCacheDiskCapacity
 	caches.SharedManager.MaxMemoryCapacity = nodeConfig.MaxCacheMemoryCapacity
@@ -425,7 +425,7 @@ func (this *Node) checkClusterConfig() error {
 		return err
 	}
 
-	logs.Println("[NODE]registering node ...")
+	logs.Println("[NODE]registering node to cluster ...")
 	resp, err := rpcClient.NodeRPC().RegisterClusterNode(rpcClient.ClusterContext(config.ClusterId, config.Secret), &pb.RegisterClusterNodeRequest{Name: HOSTNAME})
 	if err != nil {
 		return err

internal/ttlcache/cache.go

@@ -24,7 +24,7 @@ type Cache struct {
 
 func NewCache(opt ...OptionInterface) *Cache {
 	countPieces := 128
-	maxItems := 1_000_000
+	maxItems := 10_000_000
 	for _, option := range opt {
 		if option == nil {
 			continue

internal/ttlcache/cache_test.go

@@ -65,6 +65,7 @@ func TestCache_Read(t *testing.T) {
 	for i := 0; i < 10_000_000; i++ {
 		cache.Write("HELLO_WORLD_"+strconv.Itoa(i), i, time.Now().Unix()+int64(i%10240)+1)
 	}
+	time.Sleep(10 * time.Second)
 
 	total := 0
 	for _, piece := range cache.pieces {

internal/utils/reader_utils.go

@@ -0,0 +1,25 @@
+// Copyright 2021 Liuxiangchao iwind.liu@gmail.com. All rights reserved.
+
+package utils
+
+import "io"
+
+func CopyWithFilter(writer io.Writer, reader io.Reader, buf []byte, filter func(p []byte) []byte) (written int64, err error) {
+	for {
+		n, err := reader.Read(buf)
+		if n > 0 {
+			n2, err := writer.Write(filter(buf[:n]))
+			written += int64(n2)
+			if err != nil {
+				return written, err
+			}
+		}
+		if err != nil {
+			if err == io.EOF {
+				break
+			}
+			return written, err
+		}
+	}
+	return written, nil
+}