b612
/
starmap
1
0
Fork 0
Code Issues Pull Requests Projects Releases 5 Wiki Activity

Compare commits

base: b612:master
Branches Tags
b612:master
b612:v1.2.4
b612:v1.2.3
b612:v1.2.2
b612:v1.2.1
b612:v1.2.0
b612:v1.1.1
b612:v1.1.0
b612:v1.0.0
..
compare: b612:v1.1.0
Branches Tags
b612:master
b612:v1.2.4
b612:v1.2.3
b612:v1.2.2
b612:v1.2.1
b612:v1.2.0
b612:v1.1.1
b612:v1.1.0
b612:v1.0.0

No commits in common. 'master' and 'v1.1.0' have entirely different histories.
master ... v1.1.0

15 changed files with 384 additions and 655 deletions
Show Stats

75
basicrw.go
Unescape Escape View File

@ -1,24 +1,21 @@
package starmap package starmap
import ( import "errors"
"errors"
"os"
)
func (stack *StarStackMem) Count() int { func (stack *StarStack) Count() int {
stack.kvPushmu.Lock() stack.kvPushmu.Lock()
defer stack.kvPushmu.Unlock() defer stack.kvPushmu.Unlock()
return len(stack.kvStack) return len(stack.kvStack)
} }
func (stack *StarStackMem) Push(val interface{}) error { func (stack *StarStack) Push(val interface{}) error {
stack.kvPushmu.Lock() stack.kvPushmu.Lock()
defer stack.kvPushmu.Unlock() defer stack.kvPushmu.Unlock()
stack.kvStack = append(stack.kvStack, val) stack.kvStack = append(stack.kvStack, val)
return nil return nil
} }
func (stack *StarStackMem) Pop() (interface{}, error) { func (stack *StarStack) Pop() (interface{}, error) {
stack.kvPushmu.Lock() stack.kvPushmu.Lock()
defer stack.kvPushmu.Unlock() defer stack.kvPushmu.Unlock()
if len(stack.kvStack) == 0 { if len(stack.kvStack) == 0 {
@ -29,71 +26,43 @@ func (stack *StarStackMem) Pop() (interface{}, error) {
return val, nil return val, nil
} }
func (stack *StarStackMem) MustPop() interface{} { func (stack *StarStack) MustPop() interface{} {
val, _ := stack.Pop() val, _ := stack.Pop()
return val return val
} }
func Get(key interface{}) (interface{}, error) { func Get(key string) (interface{}, error) {
return globalMap.Get(key)
}
func (m *StarMapKV) Get(key interface{}) (interface{}, error) {
var err error var err error
m.mu.RLock() kvmu.RLock()
defer m.mu.RUnlock() defer kvmu.RUnlock()
data, ok := m.kvMap[key] data, ok := kvMap[key]
if !ok { if !ok {
err = os.ErrNotExist err = errors.New("key not exists")
} }
return data, err return data, err
} }
func (m *StarMapKV) MustGet(key interface{}) interface{} { func MustGet(key string) interface{} {
result, _ := m.Get(key) result, _ := Get(key)
return result return result
} }
func MustGet(key interface{}) interface{} {
return globalMap.MustGet(key)
}
func Store(key interface{}, value interface{}) error { func Store(key string, value interface{}) error {
return globalMap.Store(key, value) kvmu.Lock()
} defer kvmu.Unlock()
func (m *StarMapKV) Store(key interface{}, value interface{}) error { kvMap[key] = value
m.mu.Lock()
defer m.mu.Unlock()
m.kvMap[key] = value
return nil return nil
} }
func Exists(key interface{}) bool { func Delete(key string) error {
return globalMap.Exists(key) kvmu.Lock()
} defer kvmu.Unlock()
delete(kvMap, key)
func (m *StarMapKV) Exists(key interface{}) bool {
m.mu.RLock()
defer m.mu.RUnlock()
_, ok := m.kvMap[key]
return ok
}
func Delete(key interface{}) error {
return globalMap.Delete(key)
}
func (m *StarMapKV) Delete(key interface{}) error {
m.mu.Lock()
defer m.mu.Unlock()
delete(m.kvMap, key)
return nil return nil
} }
func Range(run func(k interface{}, v interface{}) bool) error { func Range(run func(k string, v interface{}) bool) error {
return globalMap.Range(run) for k, v := range kvMap {
}
func (m *StarMapKV) Range(run func(k interface{}, v interface{}) bool) error {
for k, v := range m.kvMap {
if !run(k, v) { if !run(k, v) {
break break
} }

97
client_basicrw.go
Unescape Escape View File

@ -1,11 +1,19 @@
package starmap package starmap
import ( import (
"b612.me/notify" "errors"
"time"
) )
func (kv *RemoteKv) clientOnline() error { func (kv *RemoteKv) clientOnline() error {
return kv.reconnect() if kv.client == nil || (kv.setKeepAlive && !kv.client.Online) {
return errors.New("Client Not Online,Please Contact A Server First")
}
if (!kv.setKeepAlive) && (!kv.client.Online) {
err := kv.reconnect()
return err
}
return nil
} }
func (kv *RemoteKv) MustGet(key string) interface{} { func (kv *RemoteKv) MustGet(key string) interface{} {
@ -13,84 +21,77 @@ func (kv *RemoteKv) MustGet(key string) interface{} {
return result return result
} }
func (kv *RemoteKv) Get(key interface{}) (interface{}, error) { func (kv *RemoteKv) Get(key string) (interface{}, error) {
kv.kvmu.Lock()
defer kv.kvmu.Unlock()
defer kv.keepalive()
if err := kv.clientOnline(); err != nil { if err := kv.clientOnline(); err != nil {
return nil, err return nil, err
} }
keyCode, err := notify.ToMsgVal(key) data, err := kv.client.SendValueWait("Get", key, kv.TimeOut)
if err != nil { if err != nil {
return nil, err return nil, err
} }
data, err := kv.client.SendWait("get", keyCode, kv.readTimeout) inf, err := decode([]byte(data.Value))
if err != nil { if err != nil {
return nil, err return nil, err
} }
rp, err := data.Value.ToInterface() var repInf *basicReply
if err != nil { repInf = inf.(*basicReply)
return nil, err if repInf.Key == "ok" {
inf2, err := decode(repInf.Value)
if err != nil {
return nil, err
}
return inf2, nil
} }
reply := rp.(kvMsg) return nil, errors.New(string(repInf.Value))
return reply.Value, reply.Err
} }
func (kv *RemoteKv) Store(key interface{}, value interface{}) error { func (kv *RemoteKv) Store(key string, value interface{}) error {
kv.kvmu.Lock()
defer kv.kvmu.Unlock()
defer kv.keepalive()
if err := kv.clientOnline(); err != nil { if err := kv.clientOnline(); err != nil {
return err return err
} }
encodeData, err := notify.ToMsgVal(kvMsg{ encodeData, err := encode(value)
Key: key,
Value: value,
Err: nil,
})
if err != nil { if err != nil {
return err return err
} }
data, err := kv.client.SendWait("store", encodeData, kv.readTimeout) sendData, _ := encode(basicReply{key, encodeData})
data, err := kv.client.SendValueWait("Store", string(sendData), kv.TimeOut)
if err != nil { if err != nil {
return err return err
} }
rp, err := data.Value.ToInterface() if data.Value == "ok" {
if err != nil { return nil
return err
} }
return rp.(kvMsg).Err return errors.New(data.Value)
} }
func (kv *RemoteKv) Delete(key interface{}) error { func (kv *RemoteKv) Delete(key string) error {
kv.kvmu.Lock()
defer kv.kvmu.Unlock()
defer kv.keepalive()
if err := kv.clientOnline(); err != nil { if err := kv.clientOnline(); err != nil {
return err return err
} }
keyCode, err := notify.ToMsgVal(key) data, err := kv.client.SendValueWait("Delete", key, kv.TimeOut)
if err != nil { if err != nil {
return err return err
} }
data, err := kv.client.SendWait("delete", keyCode, kv.readTimeout) if data.Value == "ok" {
if err != nil { return nil
return err
} }
rp, err := data.Value.ToInterface() return errors.New(data.Value)
if err != nil {
return err
}
return rp.(kvMsg).Err
} }
func (kv *RemoteKv) Exists(key interface{}) bool { func (kv *RemoteKv) keepalive() {
if err := kv.clientOnline(); err != nil { if !kv.setKeepAlive && kv.client.Online {
return false kv.client.ClientStop()
} for kv.client.Online {
keyCode, err := notify.ToMsgVal(key) time.Sleep(time.Millisecond)
if err != nil { }
return false
}
data, err := kv.client.SendWait("exists", keyCode, kv.readTimeout)
if err != nil {
return false
}
rp, err := data.Value.ToInterface()
if err != nil {
return false
} }
reply := rp.(kvMsg)
return reply.Value.(bool)
} }

55
client_prorw.go
Unescape Escape View File

@ -0,0 +1,55 @@
package starmap
import (
"errors"
)
func (kv *RemoteKv) GetMap(key string) (StarMap, error) {
kv.kvmu.Lock()
defer kv.kvmu.Unlock()
defer kv.keepalive()
recv, err := kv.client.SendValueWait("GetMap", key, kv.TimeOut)
if err != nil {
return StarMapSample{}, err
}
if recv.Key == "error" {
return StarMapSample{}, errors.New(recv.Value)
}
value, err := decodeMap([]byte(recv.Value))
if err != nil {
return StarMapSample{}, err
}
return value, nil
}
func (kv *RemoteKv) StoreMap(value StarMap) error {
kv.kvmu.Lock()
defer kv.kvmu.Unlock()
defer kv.keepalive()
encodeData, err := encodeMap(value)
if err != nil {
return err
}
recv, err := kv.client.SendValueWait("StoreMap", string(encodeData), kv.TimeOut)
if err != nil {
return err
}
if recv.Key == "error" {
return errors.New(recv.Value)
}
return nil
}
func (kv *RemoteKv) DeleteMap(value StarMap) error {
kv.kvmu.Lock()
defer kv.kvmu.Unlock()
defer kv.keepalive()
recv, err := kv.client.SendValueWait("DeleteMap", value.GetName(), kv.TimeOut)
if err != nil {
return err
}
if recv.Key == "error" {
return errors.New(recv.Value)
}
return nil
}

5
go.mod
Unescape Escape View File

@ -1,5 +0,0 @@
module b612.me/starmap
go 1.16
require b612.me/notify v0.0.0-20240818092352-85803f75dfa0

76
go.sum
Unescape Escape View File

@ -1,76 +0,0 @@
b612.me/notify v0.0.0-20240818092352-85803f75dfa0 h1:PaLuNLMM0HBM7qPdk4igtNvqT2CDgdm52sL+KA9I3so=
b612.me/notify v0.0.0-20240818092352-85803f75dfa0/go.mod h1:YyaYF4jj5ygIC/QbCMyhBWnsmmTL8kVs6UlHMGVKiY8=
b612.me/starcrypto v0.0.5 h1:Aa4pRDO2lBH2Aw+vz8NuUtRb73J8z5aOa9SImBY5sq4=
b612.me/starcrypto v0.0.5/go.mod h1:pF5A16p8r/h1G0x7ZNmmAF6K1sdIMpbCUxn2WGC8gZ0=
b612.me/stario v0.0.0-20240818091810-d528a583f4b2 h1:SxN1WDZsEBQFTnLaKbc7Z+91uyWhUB4cKHo5Ucztyh0=
b612.me/stario v0.0.0-20240818091810-d528a583f4b2/go.mod h1:1Owmu9jzKWgs4VsmeI8YWlGwLrCwPNM/bYpxkyn+MMk=
github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY=
golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
golang.org/x/crypto v0.13.0/go.mod h1:y6Z2r+Rw4iayiXXAIxJIDAJ1zMW4yaTpebo8fPOliYc=
golang.org/x/crypto v0.19.0/go.mod h1:Iy9bg/ha4yyC70EfRS8jz+B6ybOBKMaSxLj6P6oBDfU=
golang.org/x/crypto v0.21.0/go.mod h1:0BP7YvVV9gBbVKyeTG0Gyn+gZm94bibOW5BjDEYAOMs=
golang.org/x/crypto v0.23.0/go.mod h1:CKFgDieR+mRhux2Lsu27y0fO304Db0wZe70UKqHu0v8=
golang.org/x/crypto v0.26.0 h1:RrRspgV4mU+YwB4FYnuBoKsUapNIL5cohGAmSH3azsw=
golang.org/x/crypto v0.26.0/go.mod h1:GY7jblb9wI+FOo5y8/S2oY4zWP07AkOJ4+jxCqdqn54=
golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4=
golang.org/x/mod v0.8.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs=
golang.org/x/mod v0.12.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs=
golang.org/x/mod v0.15.0/go.mod h1:hTbmBsO62+eylJbnUtE2MGJUyE7QWk4xUqPFrRgJ+7c=
golang.org/x/mod v0.17.0/go.mod h1:hTbmBsO62+eylJbnUtE2MGJUyE7QWk4xUqPFrRgJ+7c=
golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
golang.org/x/net v0.10.0/go.mod h1:0qNGK6F8kojg2nk9dLZ2mShWaEBan6FAoqfSigmmuDg=
golang.org/x/net v0.15.0/go.mod h1:idbUs1IY1+zTqbi8yxTbhexhEEk5ur9LInksu6HrEpk=
golang.org/x/net v0.21.0/go.mod h1:bIjVDfnllIU7BJ2DNgfnXvpSvtn8VRwhlsaeUTyUS44=
golang.org/x/net v0.25.0/go.mod h1:JkAGAh7GEvH74S6FOH42FLoXpXbE/aqXSrIQjXgsiwM=
golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
golang.org/x/sync v0.3.0/go.mod h1:FU7BRWz2tNW+3quACPkgCx/L+uEAv1htQ0V83Z9Rj+Y=
golang.org/x/sync v0.6.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
golang.org/x/sync v0.7.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
golang.org/x/sync v0.8.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.12.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.17.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
golang.org/x/sys v0.18.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
golang.org/x/sys v0.20.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
golang.org/x/sys v0.23.0 h1:YfKFowiIMvtgl1UERQoTPPToxltDeZfbj4H7dVUCwmM=
golang.org/x/sys v0.23.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
golang.org/x/telemetry v0.0.0-20240228155512-f48c80bd79b2/go.mod h1:TeRTkGYfJXctD9OcfyVLyj2J3IxLnKwHJR8f4D8a3YE=
golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k=
golang.org/x/term v0.8.0/go.mod h1:xPskH00ivmX89bAKVGSKKtLOWNx2+17Eiy94tnKShWo=
golang.org/x/term v0.12.0/go.mod h1:owVbMEjm3cBLCHdkQu9b1opXd4ETQWc3BhuQGKgXgvU=
golang.org/x/term v0.17.0/go.mod h1:lLRBjIVuehSbZlaOtGMbcMncT+aqLLLmKrsjNrUguwk=
golang.org/x/term v0.18.0/go.mod h1:ILwASektA3OnRv7amZ1xhE/KTR+u50pbXfZ03+6Nx58=
golang.org/x/term v0.20.0/go.mod h1:8UkIAJTvZgivsXaD6/pH6U9ecQzZ45awqEOzuCvwpFY=
golang.org/x/term v0.23.0 h1:F6D4vR+EHoL9/sWAWgAR1H2DcHr4PareCbAaCo1RpuU=
golang.org/x/term v0.23.0/go.mod h1:DgV24QBUrK6jhZXl+20l6UWznPlwAHm1Q1mGHtydmSk=
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
golang.org/x/text v0.9.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8=
golang.org/x/text v0.13.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE=
golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
golang.org/x/text v0.15.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
golang.org/x/text v0.17.0/go.mod h1:BuEKDfySbSR4drPmRPG/7iBdf8hvFMuRexcpahXilzY=
golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc=
golang.org/x/tools v0.6.0/go.mod h1:Xwgl3UAJ/d3gWutnCtw505GrjyAbvKui8lOU390QaIU=
golang.org/x/tools v0.13.0/go.mod h1:HvlwmtVNQAhOuCjW7xxvovg8wbNq7LwfXh/k7wXUl58=
golang.org/x/tools v0.21.1-0.20240508182429-e35e4ccd0d2d/go.mod h1:aiJjzUbINMkxbQROHiO6hDPo2LHcIPhhQsa9DLh0yGk=
golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=

37
prorw.go
Unescape Escape View File

@ -0,0 +1,37 @@
package starmap
import (
"errors"
)
type StarMap interface {
GetName() string
}
type StarMapSample struct{}
func (sample StarMapSample) GetName() string {
return "starsample"
}
func GetMap(value StarMap) error {
kvmuPro.RLock()
defer kvmuPro.RUnlock()
value, ok := kvMapPro[value.GetName()]
if !ok {
return errors.New("key not exists")
}
return nil
}
func StoreMap(value StarMap) error {
kvmuPro.Lock()
defer kvmuPro.Unlock()
kvMapPro[value.GetName()] = value
return nil
}
func DeleteMap(value StarMap) error {
delete(kvMapPro, value.GetName())
return nil
}

113
remote_define.go
Unescape Escape View File

@ -2,73 +2,45 @@ package starmap
import ( import (
"encoding/gob" "encoding/gob"
"os" "errors"
"sync"
"time" "time"
"b612.me/notify" "b612.me/notify"
) )
func init() {
notify.RegisterName("b612.me/starmap/kvmsg", kvMsg{})
notify.RegisterName("b612.me/starmap/error", starMapErr{})
}
type starMapErr struct {
Err string
}
func (s starMapErr) Error() string {
return s.Err
}
func newStarMapErr(err error) error {
if err == nil {
return nil
}
return starMapErr{Err: err.Error()}
}
type kvMsg struct {
Key interface{}
Value interface{}
Err error
}
type RemoteKv struct { type RemoteKv struct {
server notify.Server kvmu sync.RWMutex
client notify.Client server *notify.StarNotifyS
kvmap StarMapKV client *notify.StarNotifyC
addr string addr string
network string setKeepAlive bool
readTimeout time.Duration TimeOut time.Duration
timeout time.Duration ErrMsg string
} }
func NewServer(network, addr string) (*RemoteKv, error) { func NewServer(addr string) (*RemoteKv, error) {
var kv RemoteKv
var err error var err error
kv := RemoteKv{ kv.addr = addr
server: notify.NewServer(), kv.TimeOut = time.Second * 10
kvmap: NewStarMap(), kv.server, err = notify.NewNotifyS("tcp", addr)
addr: addr,
network: network,
}
err = kv.server.Listen(network, addr)
if err == nil { if err == nil {
kv.bind() kv.bind()
} }
return &kv, err return &kv, err
} }
func NewClient(network, addr string, dialTimeout time.Duration) (*RemoteKv, error) { func NewClient(addr string) (*RemoteKv, error) {
var err error var err error
kv := RemoteKv{ var kv RemoteKv
client: notify.NewClient(), kv.addr = addr
kvmap: NewStarMap(), kv.TimeOut = time.Second * 10
addr: addr, kv.client, err = notify.NewNotifyC("tcp", addr)
network: network, if err == nil {
timeout: dialTimeout, kv.setKeepAlive = true
readTimeout: time.Second * 5, kv.client.UseChannel = false
} }
err = kv.client.ConnectTimeout(network, addr, dialTimeout)
return &kv, err return &kv, err
} }
@ -82,21 +54,38 @@ func (kv *RemoteKv) RegisterAll(data []interface{}) {
} }
} }
func (kv *RemoteKv) bind() { func (kv *RemoteKv) SetKeepAlive(alive bool) {
//for server kv.setKeepAlive = alive
kv.server.SetDefaultLink(kv.dispatch) }
//for client
func (kv *RemoteKv) GetKeepAlive() bool {
return kv.setKeepAlive
} }
func (kv *RemoteKv) reconnect() error { func (kv *RemoteKv) reconnect() error {
if kv.server != nil { var err error
return nil if kv.addr != "" {
} kv.client, err = notify.NewNotifyC("tcp", kv.addr)
if kv.client != nil { if err == nil {
if kv.client.Status().Alive { kv.client.UseChannel = false
return nil
} }
return kv.client.ConnectTimeout(kv.network, kv.addr, kv.timeout) return err
}
return errors.New("Client Not Online!")
}
func (kv *RemoteKv) bind() {
kv.server.SetNotify("Get", kv.get)
kv.server.SetNotify("Store", kv.store)
kv.server.SetNotify("Delete", kv.store)
kv.server.SetNotify("GetMap", kv.getMap)
kv.server.SetNotify("StoreMap", kv.storeMap)
kv.server.SetNotify("DeleteMap", kv.deleteMap)
}
func (kv *RemoteKv) ErrTimeOut(err error) bool {
if err.Error() == "Time Exceed" {
return true
} }
return os.ErrInvalid return false
} }

92
server_basicrw.go
Unescape Escape View File

@ -1,39 +1,63 @@
package starmap package starmap
import ( import "b612.me/notify"
"b612.me/notify"
)
func (r *RemoteKv) dispatch(msg *notify.Message) { type basicReply struct {
switch msg.Key { Key string
case "get": Value []byte
data, err := r.kvmap.Get(msg.Value.MustToInterface()) }
msg.ReplyObj(kvMsg{
Key: msg.Value.MustToInterface(), func (kv *RemoteKv) get(data notify.SMsg) string {
Value: data, key := data.Value
Err: newStarMapErr(err), value, err := Get(key)
}) if err != nil {
case "delete": sendData, err := encode(basicReply{"error", []byte("key not exists")})
err := r.kvmap.Delete(msg.Value.MustToInterface()) if err != nil {
msg.ReplyObj(kvMsg{ kv.ErrMsg += err.Error()
Key: msg.Value.MustToInterface(), return ""
Value: nil, }
Err: newStarMapErr(err), data.Reply(string(sendData))
}) } else {
case "exists": prepareData, err := encode(value)
ext := r.kvmap.Exists(msg.Value.MustToInterface()) if err != nil {
msg.ReplyObj(kvMsg{ kv.ErrMsg += err.Error()
Key: msg.Value.MustToInterface(), sendData, _ := encode(basicReply{"error", []byte(err.Error())})
Value: ext, data.Reply(string(sendData))
Err: newStarMapErr(nil), return ""
}) }
case "store": sendData, err := encode(basicReply{"ok", prepareData})
ext := msg.Value.MustToInterface().(kvMsg) if err != nil {
err := r.kvmap.Store(ext.Key, ext.Value) kv.ErrMsg += err.Error()
msg.ReplyObj(kvMsg{ sendData, _ := encode(basicReply{"error", []byte(err.Error())})
Key: msg.Value.MustToInterface(), data.Reply(string(sendData))
Value: nil, return ""
Err: newStarMapErr(err), }
}) data.Reply(string(sendData))
}
return ""
}
func (kv *RemoteKv) store(data notify.SMsg) string {
decodeData, err := decode([]byte(data.Value))
if err != nil {
return err.Error()
}
inf := decodeData.(*basicReply)
customStruct, err := decode(inf.Value)
if err != nil {
return err.Error()
}
err = Store(inf.Key, customStruct)
if err != nil {
return err.Error()
}
return "ok"
}
func (kv *RemoteKv) delete(data notify.SMsg) string {
err := Delete(data.Value)
if err != nil {
return err.Error()
} }
return "ok"
} }

42
server_prorw.go
Unescape Escape View File

@ -0,0 +1,42 @@
package starmap
import (
"b612.me/notify"
)
func (kv *RemoteKv) getMap(data notify.SMsg) string {
val, ok := kvMapPro[data.Value]
if !ok {
data.Key = "error"
data.Reply("key not exists")
return ""
}
encodeData, err := encodeMap(val)
if err != nil {
data.Key = "error"
data.Reply(err.Error())
return ""
}
return string(encodeData)
}
func (kv *RemoteKv) storeMap(data notify.SMsg) string {
recvData, err := decodeMap([]byte(data.Value))
if err != nil {
data.Key = "error"
data.Reply(err.Error())
return ""
}
err = StoreMap(recvData)
if err != nil {
data.Key = "error"
data.Reply(err.Error())
return ""
}
return "ok"
}
func (kv *RemoteKv) deleteMap(data notify.SMsg) string {
delete(kvMapPro, data.Value)
return "ok"
}

282
stack.go
Unescape Escape View File

@ -1,282 +0,0 @@
package starmap
import (
"errors"
"fmt"
"io"
"os"
"runtime"
"sync"
"sync/atomic"
"time"
)
type StarStack struct {
datas []interface{}
pStart uint64
pEnd uint64
cap uint64
isClose atomic.Value
rmu sync.Mutex
wmu sync.Mutex
}
func NewStarStack(cap uint64) *StarStack {
rtnBuffer := new(StarStack)
rtnBuffer.cap = cap
rtnBuffer.datas = make([]interface{}, cap)
rtnBuffer.isClose.Store(false)
return rtnBuffer
}
func (star *StarStack) init() {
star.cap = 1024
star.datas = make([]interface{}, star.cap)
star.isClose.Store(false)
}
func (star *StarStack) Free() uint64 {
return star.cap - star.Len()
}
func (star *StarStack) Cap() uint64 {
return star.cap
}
func (star *StarStack) Len() uint64 {
if star.pEnd >= star.pStart {
return star.pEnd - star.pStart
}
return star.pEnd - star.pStart + star.cap
}
func (star *StarStack) PopNoWait() (interface{}, error) {
if star.isClose.Load() == nil {
star.init()
}
if star.isClose.Load().(bool) {
return 0, io.EOF
}
if star.Len() == 0 {
return 0, os.ErrNotExist
}
nowPtr := star.pStart
nextPtr := star.pStart + 1
if nextPtr >= star.cap {
nextPtr = 0
}
data := star.datas[nowPtr]
ok := atomic.CompareAndSwapUint64(&star.pStart, nowPtr, nextPtr)
if !ok {
return 0, os.ErrInvalid
}
return data, nil
}
func (star *StarStack) MustPop() interface{} {
if star.isClose.Load() == nil {
star.init()
}
data, err := star.Pop()
if err != nil {
return nil
}
return data
}
func (star *StarStack) Pop() (interface{}, error) {
if star.isClose.Load() == nil {
star.init()
}
for {
if star.isClose.Load().(bool) {
return 0, io.EOF
}
if star.Len() == 0 {
return 0, os.ErrNotExist
}
nowPtr := star.pStart
nextPtr := star.pStart + 1
if nextPtr >= star.cap {
nextPtr = 0
}
data := star.datas[nowPtr]
ok := atomic.CompareAndSwapUint64(&star.pStart, nowPtr, nextPtr)
if !ok {
time.Sleep(time.Microsecond)
runtime.Gosched()
continue
}
return data, nil
}
}
func (star *StarStack) Push(data interface{}) error {
if star.isClose.Load() == nil {
star.init()
}
if star.isClose.Load().(bool) {
return io.EOF
}
nowPtr := star.pEnd
kariEnd := nowPtr + 1
if kariEnd == star.cap {
kariEnd = 0
}
if kariEnd == atomic.LoadUint64(&star.pStart) {
for {
time.Sleep(time.Microsecond)
runtime.Gosched()
if kariEnd != atomic.LoadUint64(&star.pStart) {
break
}
}
}
star.datas[nowPtr] = data
if ok := atomic.CompareAndSwapUint64(&star.pEnd, nowPtr, kariEnd); !ok {
return os.ErrInvalid
}
return nil
}
func (star *StarStack) Close() error {
if star.isClose.Load() == nil {
star.init()
}
star.isClose.Store(true)
return nil
}
func (star *StarStack) Read(buf []interface{}) (int, error) {
if star.isClose.Load() == nil {
star.init()
}
if star.isClose.Load().(bool) {
return 0, io.EOF
}
if buf == nil {
return 0, errors.New("buffer is nil")
}
star.rmu.Lock()
defer star.rmu.Unlock()
var sum int = 0
for i := 0; i < len(buf); i++ {
data, err := star.PopNoWait()
if err != nil {
if err == io.EOF {
return sum, err
}
if err == os.ErrNotExist {
i--
continue
}
return sum, nil
}
buf[i] = data
sum++
}
return sum, nil
}
func (star *StarStack) Write(bts []byte) (int, error) {
if star.isClose.Load() == nil {
star.init()
}
if bts == nil || star.isClose.Load().(bool) {
return 0, io.EOF
}
star.wmu.Lock()
defer star.wmu.Unlock()
var sum = 0
for i := 0; i < len(bts); i++ {
err := star.Push(bts[i])
if err != nil {
fmt.Println("Write bts err:", err)
return sum, err
}
sum++
}
return sum, nil
}
type StarChanStack struct {
data chan interface{}
cap uint64
current uint64
isClose atomic.Value
}
func NewStarChanStack(cap uint64) *StarChanStack {
rtnBuffer := new(StarChanStack)
rtnBuffer.cap = cap
rtnBuffer.isClose.Store(false)
rtnBuffer.data = make(chan interface{}, cap)
return rtnBuffer
}
func (star *StarChanStack) init() {
star.cap = 1024
star.data = make(chan interface{}, star.cap)
star.isClose.Store(false)
}
func (star *StarChanStack) Free() uint64 {
return star.cap - star.current
}
func (star *StarChanStack) Cap() uint64 {
return star.cap
}
func (star *StarChanStack) Len() uint64 {
return star.current
}
func (star *StarChanStack) Pop() (interface{}, error) {
if star.isClose.Load() == nil {
star.init()
}
if star.isClose.Load().(bool) {
return 0, io.EOF
}
data, ok := <-star.data
if !ok {
star.isClose.Store(true)
return 0, errors.New("channel read error")
}
for {
current := atomic.LoadUint64(&star.current)
if atomic.CompareAndSwapUint64(&star.current, current, current-1) {
break
}
}
return data, nil
}
func (star *StarChanStack) Push(data interface{}) error {
defer func() {
recover()
}()
if star.isClose.Load() == nil {
star.init()
}
if star.isClose.Load().(bool) {
return io.EOF
}
star.data <- data
for {
current := atomic.LoadUint64(&star.current)
if atomic.CompareAndSwapUint64(&star.current, current, current+1) {
break
}
}
return nil
}
func (star *StarChanStack) Close() error {
if star.isClose.Load() == nil {
star.init()
}
star.isClose.Store(true)
close(star.data)
return nil
}

66
stack_test.go
Unescape Escape View File

@ -1,66 +0,0 @@
package starmap
import (
"fmt"
"sync/atomic"
"testing"
"time"
)
func Test_Circle_Speed(t *testing.T) {
buf := StarStack{}
count := uint64(0)
for i := 1; i <= 10; i++ {
go func() {
for {
buf.Push('a')
}
}()
}
for i := 1; i <= 10; i++ {
go func() {
for {
_, err := buf.Pop()
if err == nil {
atomic.AddUint64(&count, 1)
}
}
}()
}
time.Sleep(time.Second * 10)
fmt.Println(count)
}
func Test_Chan_Circle_Speed(t *testing.T) {
buf := StarChanStack{}
count := uint64(0)
for i := 1; i <= 10; i++ {
go func() {
for {
err := buf.Push('a')
if err != nil {
fmt.Println("finished write")
break
}
}
}()
}
for i := 1; i <= 10; i++ {
go func() {
for {
_, err := buf.Pop()
if err == nil {
atomic.AddUint64(&count, 1)
} else {
fmt.Println("finished read")
break
}
}
}()
}
time.Sleep(time.Second * 10)
fmt.Println(count)
buf.Close()
time.Sleep(time.Second * 3)
}

21
starmap.go
Unescape Escape View File

@ -2,24 +2,17 @@ package starmap
import "sync" import "sync"
var globalMap StarMapKV var kvMap map[string]interface{}
var kvmu sync.RWMutex
var kvMapPro map[string]StarMap
var kvmuPro sync.RWMutex
type StarMapKV struct { type StarStack struct {
kvMap map[interface{}]interface{}
mu sync.RWMutex
}
type StarStackMem struct {
kvPushmu sync.RWMutex kvPushmu sync.RWMutex
kvStack []interface{} kvStack []interface{}
} }
func init() { func init() {
globalMap = NewStarMap() kvMap = make(map[string]interface{})
} kvMapPro = make(map[string]StarMap)
func NewStarMap() StarMapKV {
var mp StarMapKV
mp.kvMap = make(map[interface{}]interface{})
return mp
} }

38
starmap_test.go
Unescape Escape View File

@ -8,28 +8,38 @@ import (
) )
type Miaomiao struct { type Miaomiao struct {
Val1 string Sakura string
Val2 int Fuck int
Val3 bool Mimi bool
} }
func Test_Remote(t *testing.T) { func Test_Remote(t *testing.T) {
Store("test", 22222) Store("nmb", 22222)
server, _ := NewServer("tcp", "127.0.0.1:45678") server, err := NewServer("127.0.0.1:45678")
server.Register(&Miaomiao{}) server.Register(&Miaomiao{})
client, _ := NewClient("tcp", "127.0.0.1:45678", time.Second*2) fmt.Println(err)
client, err := NewClient("127.0.0.1:45678")
client.SetKeepAlive(false)
fmt.Println(err)
_ = server _ = server
fmt.Println(client.Get("meow")) fmt.Println(client.Get("nmb"))
fmt.Println(client.Exists("meow")) fmt.Println(client.Store("maio", Miaomiao{"sss", 222, true}))
fmt.Println(client.Store("meow", Miaomiao{"sss", 222, true})) fmt.Println(client.Get("maio"))
fmt.Println(client.Get("meow"))
fmt.Println(client.Exists("meow"))
fmt.Println(client.Delete("meow"))
fmt.Println(client.Exists("meow"))
} }
func (cat *Miaomiao) GetName() string { func (cat *Miaomiao) GetName() string {
return "meow" return "miaomiao"
}
func Test_Remote2(t *testing.T) {
server, err := NewServer("127.0.0.1:45678")
server.Register(&Miaomiao{})
fmt.Println(err)
client, err := NewClient("127.0.0.1:45678")
//client.SetKeepAlive(false)
fmt.Println(err)
_ = server
fmt.Println(client.StoreMap(&Miaomiao{"suki", 1127, false}))
fmt.Println(client.GetMap("miaomiao"))
} }
func Test_Math(t *testing.T) { func Test_Math(t *testing.T) {

2
sync.go
Unescape Escape View File

@ -25,7 +25,7 @@ func (swg *WaitGroup) Add(delta int) {
} }
for { for {
allC := atomic.LoadUint32(&swg.allCount) allC := atomic.LoadUint32(&swg.allCount)
if atomic.LoadUint32(&swg.maxCount) == 0 || atomic.LoadUint32(&swg.maxCount) >= allC+uint32(delta) { if atomic.LoadUint32(&swg.maxCount) >= allC+uint32(delta) {
if delta < 0 { if delta < 0 {
atomic.AddUint32(&swg.allCount, ^uint32(Udelta)) atomic.AddUint32(&swg.allCount, ^uint32(Udelta))
} else { } else {

38
transfer.go
Unescape Escape View File

@ -0,0 +1,38 @@
package starmap
import (
"bytes"
"encoding/gob"
)
func init() {
gob.Register(&basicReply{})
}
func encode(src interface{}) ([]byte, error) {
var buf bytes.Buffer
enc := gob.NewEncoder(&buf)
err := enc.Encode(&src)
return buf.Bytes(), err
}
func decode(src []byte) (interface{}, error) {
dec := gob.NewDecoder(bytes.NewReader(src))
var dst interface{}
err := dec.Decode(&dst)
return dst, err
}
func encodeMap(src StarMap) ([]byte, error) {
var buf bytes.Buffer
enc := gob.NewEncoder(&buf)
err := enc.Encode(&src)
return buf.Bytes(), err
}
func decodeMap(src []byte) (StarMap, error) {
dec := gob.NewDecoder(bytes.NewReader(src))
var dst StarMap
err := dec.Decode(&dst)
return dst, err
}
Write Preview
Loading…
Cancel
Save