feat: online user report;log optimization;traffic statistical optimiz…

…ation;

.gitignore

@@ -1,4 +1,5 @@
 bin/
 server
 server.exe
-config.json
+config.json
+*.test.exe

ciphers/ciphers_test.go

@@ -9,7 +9,7 @@ import (
 	"testing"
 	"time"
 
-	"github.com/rc452860/vnet/log"
+	"github.com/rc452860/vnet/comm/log"
 	"github.com/rc452860/vnet/utils/datasize"
 )
 

ciphers/ssaead/cipher_conn.go

@@ -11,7 +11,7 @@ import (
 	"crypto/md5"
 	"crypto/rand"
 
-	"github.com/rc452860/vnet/log"
+	"github.com/rc452860/vnet/comm/log"
 	"github.com/rc452860/vnet/pool"
 	"golang.org/x/crypto/hkdf"
 )

ciphers/ssstream/cipher_conn.go

@@ -7,7 +7,7 @@ import (
 	"io"
 
 	connect "github.com/rc452860/vnet/conn"
-	"github.com/rc452860/vnet/log"
+	"github.com/rc452860/vnet/comm/log"
 	"github.com/rc452860/vnet/pool"
 )
 

cmd/server/server.go

@@ -11,7 +11,7 @@ import (
 
 	"github.com/rc452860/vnet/config"
 	"github.com/rc452860/vnet/db"
-	"github.com/rc452860/vnet/log"
+	"github.com/rc452860/vnet/comm/log"
 	"github.com/rc452860/vnet/proxy/server"
 	"github.com/rc452860/vnet/utils/datasize"
 )

cmd/test/test.go

@@ -2,13 +2,14 @@ package main
 
 import (
 	"fmt"
+	"net"
 )
 
 type Test struct {
 	Name string
 }
 
 func main() {
-	a := Test{}
-	fmt.Print(a.(type))
+	listen, _ := net.Listen("tcp", "0.0.0.0:8080")
+	fmt.Print(listen.Addr().String())
 }

comm/array/time_array.go

@@ -0,0 +1,90 @@
+package array
+
+import (
+	"context"
+	"reflect"
+	"sync"
+	"time"
+)
+
+type TimeArrayElement struct {
+	Time    time.Time
+	Element interface{}
+}
+
+type TimeArray struct {
+	array     []TimeArrayElement
+	Size      int
+	expire    time.Duration
+	context   context.Context
+	cancel    context.CancelFunc
+	AutoClear bool
+	sync.RWMutex
+}
+
+func NewTimeArray(expire time.Duration, autoClear bool) TimeArray {
+	result := TimeArray{
+		expire:    expire,
+		AutoClear: autoClear,
+	}
+	if autoClear {
+		ctx, cancel := context.WithCancel(context.Background())
+		result.context = ctx
+		result.cancel = cancel
+		go result.autoClear()
+	}
+	return result
+}
+
+func (t *TimeArray) Close() {
+	if t.AutoClear && t.cancel != nil {
+		t.cancel()
+	}
+}
+
+func (t *TimeArray) autoClear() {
+	tick := time.Tick(t.expire)
+	for {
+		select {
+		case <-t.context.Done():
+			return
+		case <-tick:
+		}
+		t.Clear()
+	}
+}
+
+func (t *TimeArray) Add(data interface{}) {
+	t.Lock()
+	defer t.Unlock()
+
+	t.array = append(t.array, TimeArrayElement{
+		Time:    time.Now(),
+		Element: data,
+	})
+	t.Size++
+}
+
+func (t *TimeArray) Remove(data interface{}) {
+	for i, item := range t.array {
+		if reflect.DeepEqual(data, item) {
+			t.array = append(t.array[:i], t.array[i+1:]...)
+			t.Size--
+		}
+	}
+}
+
+func (t *TimeArray) Clear() {
+	now := time.Now()
+	for i, item := range t.array {
+		if now.Sub(item.Time) > t.expire {
+			t.array = append(t.array[:i], t.array[i+1:]...)
+			t.Size--
+		}
+	}
+}
+func (t *TimeArray) Range(callback func(i int, key interface{})) {
+	for i, item := range t.array {
+		callback(i, item.Element)
+	}
+}

comm/array/time_array_test.go

@@ -0,0 +1,43 @@
+package array
+
+import (
+	"testing"
+	"time"
+)
+
+func Benchmark_Time_Array(t *testing.B) {
+	timeArray := NewTimeArray(time.Second*2, true)
+	for i := 0; i < t.N; i++ {
+		timeArray.Add("123")
+	}
+}
+
+func Benchmark_Time_Array_Remove(t *testing.B) {
+	timeArray := NewTimeArray(time.Second*2, true)
+	for i := 0; i < t.N; i++ {
+		timeArray.Add("123")
+	}
+	t.ResetTimer()
+	for i := 0; i < t.N; i++ {
+		timeArray.Remove("123")
+	}
+	t.ReportAllocs()
+}
+
+func Test_Time_Array(t *testing.T) {
+	timeArray := NewTimeArray(time.Second*2, true)
+	tick := time.Tick(time.Second)
+	index := 0
+	for {
+		<-tick
+		index++
+		if index > 10 {
+			break
+		}
+
+		timeArray.Add("aaa")
+	}
+	timeArray.Range(func(i int, v interface{}) {
+		t.Log(v.(string))
+	})
+}

comm/cache/cache.go

@@ -0,0 +1,111 @@
+package cache
+
+import (
+	"runtime"
+	"sync"
+	"time"
+)
+
+// Cache store element with a expired time
+type Cache struct {
+	*cache
+}
+
+type cache struct {
+	mapping sync.Map
+	janitor *janitor
+}
+
+type element struct {
+	Expired time.Time
+	Payload interface{}
+}
+
+// Put element in Cache with its ttl
+func (c *cache) Put(key interface{}, payload interface{}, ttl time.Duration) {
+	c.mapping.Store(key, &element{
+		Payload: payload,
+		Expired: time.Now().Add(ttl),
+	})
+}
+
+// Get element in Cache, and drop when it expired
+func (c *cache) Get(key interface{}) interface{} {
+	item, exist := c.mapping.Load(key)
+	if !exist {
+		return nil
+	}
+	elm := item.(*element)
+	// expired
+	if time.Since(elm.Expired) > 0 {
+		c.mapping.Delete(key)
+		return nil
+	}
+	return elm.Payload
+}
+
+func (c *cache) Range(callback func(key, value interface{})) {
+	c.mapping.Range(func(k, v interface{}) bool {
+		elm := v.(*element)
+		if time.Since(elm.Expired) > 0 {
+			c.mapping.Delete(k)
+		} else {
+			callback(k, elm.Payload)
+		}
+		return true
+	})
+}
+
+func (c *cache) cleanup() {
+	c.mapping.Range(func(k, v interface{}) bool {
+		// key := k.(string)
+		elm := v.(*element)
+		if time.Since(elm.Expired) > 0 {
+			c.mapping.Delete(k)
+		}
+		return true
+	})
+}
+
+func (c *cache) Size() int {
+	var result int
+	c.Range(func(k, v interface{}) {
+		result++
+	})
+	return result
+}
+
+type janitor struct {
+	interval time.Duration
+	stop     chan struct{}
+}
+
+func (j *janitor) process(c *cache) {
+	ticker := time.NewTicker(j.interval)
+	for {
+		select {
+		case <-ticker.C:
+			c.cleanup()
+		case <-j.stop:
+			ticker.Stop()
+			return
+		}
+	}
+}
+
+func stopJanitor(c *Cache) {
+	c.janitor.stop <- struct{}{}
+}
+
+// New return *Cache
+func New(interval time.Duration) *Cache {
+	j := &janitor{
+		interval: interval,
+		stop:     make(chan struct{}),
+	}
+	c := &cache{janitor: j}
+	go j.process(c)
+	C := &Cache{c}
+	runtime.SetFinalizer(C, stopJanitor)
+	return C
+}

comm/cache/cache_test.go

@@ -0,0 +1,70 @@
+package cache
+
+import (
+	"runtime"
+	"testing"
+	"time"
+)
+
+func TestCache_Basic(t *testing.T) {
+	interval := 200 * time.Millisecond
+	ttl := 20 * time.Millisecond
+	c := New(interval)
+	c.Put("int", 1, ttl)
+	c.Put("string", "a", ttl)
+
+	i := c.Get("int")
+	if i.(int) != 1 {
+		t.Error("should recv 1")
+	}
+
+	s := c.Get("string")
+	if s.(string) != "a" {
+		t.Error("should recv 'a'")
+	}
+}
+
+func TestCache_TTL(t *testing.T) {
+	interval := 200 * time.Millisecond
+	ttl := 20 * time.Millisecond
+	c := New(interval)
+	c.Put("int", 1, ttl)
+
+	i := c.Get("int")
+	if i.(int) != 1 {
+		t.Error("should recv 1")
+	}
+
+	time.Sleep(ttl * 2)
+	i = c.Get("int")
+	if i != nil {
+		t.Error("should recv nil")
+	}
+}
+
+func TestCache_AutoCleanup(t *testing.T) {
+	interval := 10 * time.Millisecond
+	ttl := 15 * time.Millisecond
+	c := New(interval)
+	c.Put("int", 1, ttl)
+
+	time.Sleep(ttl * 2)
+	i := c.Get("int")
+	if i != nil {
+		t.Error("should recv nil")
+	}
+}
+
+func TestCache_AutoGC(t *testing.T) {
+	sign := make(chan struct{})
+	go func() {
+		interval := 10 * time.Millisecond
+		ttl := 15 * time.Millisecond
+		c := New(interval)
+		c.Put("int", 1, ttl)
+		sign <- struct{}{}
+	}()
+
+	<-sign
+	runtime.GC()
+}

comm/eventbus/eventbus.go

@@ -2,16 +2,13 @@ package eventbus
 
 import (
 	evbus "github.com/asaskevich/EventBus"
-	"github.com/rc452860/vnet/utils"
 )
 
 var eventBus evbus.Bus
 
+func init() {
+	eventBus = evbus.New()
+}
 func GetEventBus() evbus.Bus {
-	utils.Lock("eventbus:eventBus")
-	defer utils.UnLock("eventbus:eventBus")
-	if eventBus != nil {
-		eventBus = evbus.New()
-	}
 	return eventBus
 }