diff --git a/pkg/anonymity/qb/anonymity.go b/pkg/anonymity/qb/anonymity.go index 8af24882..cae80901 100644 --- a/pkg/anonymity/qb/anonymity.go +++ b/pkg/anonymity/qb/anonymity.go @@ -29,6 +29,7 @@ type sNode struct { fMutex sync.RWMutex fState state.IState fSettings ISettings + fHandlerF IHandlerF fLogger logger.ILogger fAdapter adapters.IAdapter fKVDatavase database.IKVDatabase @@ -40,6 +41,7 @@ type sNode struct { func NewNode( pSett ISettings, + pHandlerF IHandlerF, pLogger logger.ILogger, pAdapter adapters.IAdapter, pKVDatavase database.IKVDatabase, @@ -49,6 +51,7 @@ func NewNode( return &sNode{ fState: state.NewBoolState(), fSettings: pSett, + fHandlerF: pHandlerF, fLogger: pLogger, fAdapter: pAdapter, fKVDatavase: pKVDatavase, @@ -156,11 +159,6 @@ func (p *sNode) GetKeysContainer() layer2.IKeysContainer { return p.fKeysContainer } -func (p *sNode) HandleFunc(pHead uint32, pHandle IHandlerF) INode { - p.setRoute(pHead, pHandle) - return p -} - // Send message without response waiting. func (p *sNode) SendPayload( _ context.Context, @@ -338,15 +336,8 @@ func (p *sNode) handleRequest( pHead iHead, pBody []byte, ) { - // get function by payload head - f, ok := p.getRoute(pHead.getRoute()) - if !ok || f == nil { - p.fLogger.PushWarn(pLogBuilder.WithType(anon_logger.CLogWarnUnknownRoute)) - return - } - // response can be nil - resp, err := f(pCtx, p, pSender, pBody) + resp, err := p.fHandlerF(pCtx, p, pSender, pBody) if err != nil { p.fLogger.PushWarn(pLogBuilder.WithType(anon_logger.CLogWarnIncorrectResponse)) return @@ -459,21 +450,6 @@ func (p *sNode) storeHashIntoDatabase(pLogBuilder anon_logger.ILogBuilder, pNetM return nil } -func (p *sNode) setRoute(pHead uint32, pHandle IHandlerF) { - p.fMutex.Lock() - defer p.fMutex.Unlock() - - p.fHandleRoutes[pHead] = pHandle -} - -func (p *sNode) getRoute(pHead uint32) (IHandlerF, bool) { - p.fMutex.RLock() - defer p.fMutex.RUnlock() - - f, ok := p.fHandleRoutes[pHead] - return f, ok -} - func (p *sNode) getAction(pActionKey string) (chan []byte, bool) { p.fMutex.RLock() defer p.fMutex.RUnlock() diff --git a/pkg/anonymity/qb/anonymity_test.go b/pkg/anonymity/qb/anonymity_test.go index 723a4db4..82700e72 100644 --- a/pkg/anonymity/qb/anonymity_test.go +++ b/pkg/anonymity/qb/anonymity_test.go @@ -56,7 +56,9 @@ func TestNodeSettings(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() - node, _, _ := testRunNodeWithDB(ctx, time.Minute, "", &tsDatabase{}) + node, _, _ := testRunNodeWithDB(ctx, time.Minute, "", &tsDatabase{}, func(ctx context.Context, i INode, ik layer2.IParticipantKey, b []byte) ([]byte, error) { + return nil, nil + }) defer testFreeNodes([]INode{node}, 9) sett := node.GetSettings() @@ -107,7 +109,9 @@ func TestComplexFetchPayload(t *testing.T) { defer cancel() addresses := [2]string{testutils.TgAddrs[2], testutils.TgAddrs[3]} - nodes, privKeys := testRunNodes(ctx, t, time.Minute, addresses, 0) + nodes, privKeys := testRunNodes(ctx, t, time.Minute, addresses, 0, func(_ context.Context, _ INode, _ layer2.IParticipantKey, reqBytes []byte) ([]byte, error) { + return []byte(string(reqBytes) + " (response)"), nil + }) if nodes[0] == nil { t.Fatal("nodes is null") } @@ -150,7 +154,9 @@ func TestF2FWithoutFriends(t *testing.T) { // 3 seconds for wait addresses := [2]string{testutils.TgAddrs[10], testutils.TgAddrs[11]} - nodes, privKeys := testRunNodes(ctx, t, 3*time.Second, addresses, 1) + nodes, privKeys := testRunNodes(ctx, t, 3*time.Second, addresses, 1, func(_ context.Context, _ INode, _ layer2.IParticipantKey, reqBytes []byte) ([]byte, error) { + return []byte(string(reqBytes) + " (response)"), nil + }) if nodes[0] == nil { t.Fatal("nodes is null") } @@ -179,19 +185,14 @@ func TestFetchPayload(t *testing.T) { defer cancel() addresses := [2]string{testutils.TgAddrs[12], testutils.TgAddrs[13]} - nodes, privKeys := testRunNodes(ctx, t, time.Minute, addresses, 4) + nodes, privKeys := testRunNodes(ctx, t, time.Minute, addresses, 4, func(_ context.Context, _ INode, _ layer2.IParticipantKey, reqBytes []byte) ([]byte, error) { + return []byte(fmt.Sprintf("echo: '%s'", string(reqBytes))), nil + }) if nodes[0] == nil { t.Fatal("nodes is null") } defer testFreeNodes(nodes[:], 4) - nodes[1].HandleFunc( - tcHead, - func(_ context.Context, _ INode, _ layer2.IParticipantKey, reqBytes []byte) ([]byte, error) { - return []byte(fmt.Sprintf("echo: '%s'", string(reqBytes))), nil - }, - ) - largeBodySize := nodes[0].GetQBProcessor().GetScheme().GetPayloadLimit() - encoding.CSizeUint64 + 1 _, err := nodes[0].FetchPayload( ctx, @@ -222,23 +223,18 @@ func TestBroadcastPayload(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() + chResult := make(chan string) addresses := [2]string{testutils.TgAddrs[14], testutils.TgAddrs[15]} - nodes, privKeys := testRunNodes(ctx, t, time.Minute, addresses, 3) + nodes, privKeys := testRunNodes(ctx, t, time.Minute, addresses, 3, func(_ context.Context, _ INode, _ layer2.IParticipantKey, reqBytes []byte) ([]byte, error) { + res := fmt.Sprintf("echo: '%s'", string(reqBytes)) + go func() { chResult <- res }() + return nil, nil + }) if nodes[0] == nil { t.Fatal("nodes is null") } defer testFreeNodes(nodes[:], 3) - chResult := make(chan string) - nodes[1].HandleFunc( - tcHead, - func(_ context.Context, _ INode, _ layer2.IParticipantKey, reqBytes []byte) ([]byte, error) { - res := fmt.Sprintf("echo: '%s'", string(reqBytes)) - go func() { chResult <- res }() - return nil, nil - }, - ) - largeBodySize := nodes[0].GetQBProcessor().GetScheme().GetPayloadLimit() - encoding.CSizeUint64 + 1 err := nodes[0].SendPayload( context.Background(), @@ -276,7 +272,9 @@ func TestEnqueuePayload(t *testing.T) { defer cancel() addresses := [2]string{testutils.TgAddrs[16], testutils.TgAddrs[17]} - nodes, privKeys := testRunNodes(ctx, t, time.Minute, addresses, 8) + nodes, privKeys := testRunNodes(ctx, t, time.Minute, addresses, 8, func(_ context.Context, _ INode, _ layer2.IParticipantKey, reqBytes []byte) ([]byte, error) { + return []byte(string(reqBytes) + " (response)"), nil + }) if nodes[0] == nil { t.Fatal("nodes is null") } @@ -321,7 +319,9 @@ func TestHandleWrapper(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() - _node, _, privKey := testRunNode(ctx, time.Minute, "", 7, 0) + _node, _, privKey := testRunNode(ctx, time.Minute, "", 7, 0, func(_ context.Context, _ INode, _ layer2.IParticipantKey, _ []byte) ([]byte, error) { + return nil, errors.New("some error") //nolint:err113 + }) defer testFreeNodes([]INode{_node}, 7) node := _node.(*sNode) @@ -372,13 +372,6 @@ func TestHandleWrapper(t *testing.T) { t.Fatal(err) // works only logger } - node.HandleFunc( - 111, - func(_ context.Context, _ INode, _ layer2.IParticipantKey, _ []byte) ([]byte, error) { - return nil, errors.New("some error") //nolint:err113 - }, - ) - msg2, err := scheme.EncryptMessage( pubKey, payload.NewPayload64( @@ -445,7 +438,9 @@ func TestStoreHashWithBroadcastMessage(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() - _node, _, privKey := testRunNode(ctx, time.Minute, "", 6, 0) + _node, _, privKey := testRunNode(ctx, time.Minute, "", 6, 0, func(ctx context.Context, i INode, ik layer2.IParticipantKey, b []byte) ([]byte, error) { + return nil, nil + }) defer testFreeNodes([]INode{_node}, 6) node := _node.(*sNode) @@ -487,7 +482,9 @@ func TestRecvSendMessage(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() - _node, _, privKey := testRunNode(ctx, time.Minute, "", 5, 0) + _node, _, privKey := testRunNode(ctx, time.Minute, "", 5, 0, func(ctx context.Context, i INode, ik layer2.IParticipantKey, b []byte) ([]byte, error) { + return nil, nil + }) defer testFreeNodes([]INode{_node}, 5) node := _node.(*sNode) @@ -549,14 +546,14 @@ func TestRecvSendMessage(t *testing.T) { // nodes[2], nodes[3], nodes[4] = routes // nodes[2], nodes[4] are have open ports // Scheme: (nodes[0]) -> nodes[2] -> nodes[3] -> nodes[4] -> (nodes[1]) -func testRunNodes(ctx context.Context, t *testing.T, timeWait time.Duration, addresses [2]string, typeDB int) ([5]INode, [5]asymmetric.IPrivKey) { +func testRunNodes(ctx context.Context, t *testing.T, timeWait time.Duration, addresses [2]string, typeDB int, handlerF IHandlerF) ([5]INode, [5]asymmetric.IPrivKey) { nodes := [5]INode{} networkNodes := [5]network.INode{} privKeys := [5]asymmetric.IPrivKey{} addrs := [5]string{"", "", addresses[0], "", addresses[1]} for i := 0; i < 5; i++ { - nodes[i], networkNodes[i], privKeys[i] = testRunNode(ctx, timeWait, addrs[i], typeDB, i) + nodes[i], networkNodes[i], privKeys[i] = testRunNode(ctx, timeWait, addrs[i], typeDB, i, handlerF) if nodes[i] == nil { t.Errorf("node (%d) is not running %d", i, typeDB) return [5]INode{}, [5]asymmetric.IPrivKey{} @@ -569,16 +566,6 @@ func testRunNodes(ctx context.Context, t *testing.T, timeWait time.Duration, add nodes[0].GetKeysContainer().Add(pubKey1) nodes[1].GetKeysContainer().Add(pubKey0) - for _, node := range nodes { - node.HandleFunc( - tcHead, - func(_ context.Context, _ INode, _ layer2.IParticipantKey, reqBytes []byte) ([]byte, error) { - // send response - return []byte(string(reqBytes) + " (response)"), nil - }, - ) - } - go func() { if err := networkNodes[2].Run(ctx); err != nil && !errors.Is(err, net.ErrClosed) && !errors.Is(err, context.Canceled) { t.Error(err) @@ -645,7 +632,7 @@ func (p *stLogging) HasErro() bool { } */ -func testRunNodeWithDB(ctx context.Context, timeWait time.Duration, addr string, db database.IKVDatabase) (INode, network.INode, asymmetric.IPrivKey) { +func testRunNodeWithDB(ctx context.Context, timeWait time.Duration, addr string, db database.IKVDatabase, handlerF IHandlerF) (INode, network.INode, asymmetric.IPrivKey) { privKey := asymmetric.NewPrivKey() msgChan := make(chan layer1.IMessage) parallel := uint64(1) @@ -678,6 +665,7 @@ func testRunNodeWithDB(ctx context.Context, timeWait time.Duration, addr string, FServiceName: "TEST", FFetchTimeout: timeWait, }), + handlerF, // internal_std_logger.NewStdLogger(&stLogging{}, internal_anon_logger.GetLogFunc()), logger.NewLogger( logger.NewSettings(&logger.SSettings{}), @@ -721,12 +709,12 @@ func testRunNodeWithDB(ctx context.Context, timeWait time.Duration, addr string, return node, networkNode, privKey } -func testRunNode(ctx context.Context, timeWait time.Duration, addr string, typeDB, numDB int) (INode, network.INode, asymmetric.IPrivKey) { +func testRunNode(ctx context.Context, timeWait time.Duration, addr string, typeDB, numDB int, handlerF IHandlerF) (INode, network.INode, asymmetric.IPrivKey) { db, err := database.NewKVDatabase(fmt.Sprintf(tcPathDBTemplate, typeDB, numDB)) if err != nil { panic(err) } - return testRunNodeWithDB(ctx, timeWait, addr, db) + return testRunNodeWithDB(ctx, timeWait, addr, db, handlerF) } func testFreeNodes(nodes []INode, typeDB int) { diff --git a/pkg/anonymity/qb/examples/echo/construct_a.go b/pkg/anonymity/qb/examples/echo/construct_a.go index 8f862fd7..7bf7f284 100644 --- a/pkg/anonymity/qb/examples/echo/construct_a.go +++ b/pkg/anonymity/qb/examples/echo/construct_a.go @@ -69,6 +69,9 @@ func newNode(serviceName, address string) *sNode { FServiceName: serviceName, FFetchTimeout: time.Minute, }), + func(_ context.Context, _ anonymity.INode, _ layer2.IParticipantKey, b []byte) ([]byte, error) { + return []byte("echo: " + string(b)), nil + }, logger.NewLogger( logger.NewSettings(&logger.SSettings{ FInfo: os.Stdout, diff --git a/pkg/anonymity/qb/examples/echo/main.go b/pkg/anonymity/qb/examples/echo/main.go index 67891c5f..e3e39e06 100644 --- a/pkg/anonymity/qb/examples/echo/main.go +++ b/pkg/anonymity/qb/examples/echo/main.go @@ -5,8 +5,6 @@ import ( "fmt" "time" - anonymity "github.com/number571/go-peer/pkg/anonymity/qb" - "github.com/number571/go-peer/pkg/crypto/scheme/layer2" "github.com/number571/go-peer/pkg/payload" ) @@ -49,12 +47,6 @@ func runClientNode() *sNode { func runServiceNode() *sNode { ctx := context.Background() node := newNode("snode", nodeAddress) - node.fAnonymity.HandleFunc( - nodeRouter, - func(_ context.Context, _ anonymity.INode, _ layer2.IParticipantKey, b []byte) ([]byte, error) { - return []byte("echo: " + string(b)), nil - }, - ) go func() { _ = node.fAnonymity.Run(ctx) }() go func() { _ = node.fNetwork.Run(ctx) }() diff --git a/pkg/anonymity/qb/examples/ping-pong/construct_a.go b/pkg/anonymity/qb/examples/ping-pong/construct_a.go index 8f862fd7..8e2c316a 100644 --- a/pkg/anonymity/qb/examples/ping-pong/construct_a.go +++ b/pkg/anonymity/qb/examples/ping-pong/construct_a.go @@ -16,9 +16,11 @@ import ( "github.com/number571/go-peer/pkg/crypto/scheme/layer1" "github.com/number571/go-peer/pkg/crypto/scheme/layer2" "github.com/number571/go-peer/pkg/crypto/scheme/layer2/hybrid" + "github.com/number571/go-peer/pkg/encoding" "github.com/number571/go-peer/pkg/logger" "github.com/number571/go-peer/pkg/network" "github.com/number571/go-peer/pkg/network/conn" + "github.com/number571/go-peer/pkg/payload" "github.com/number571/go-peer/pkg/storage/cache" "github.com/number571/go-peer/pkg/storage/database" ) @@ -69,6 +71,7 @@ func newNode(serviceName, address string) *sNode { FServiceName: serviceName, FFetchTimeout: time.Minute, }), + handler, logger.NewLogger( logger.NewSettings(&logger.SSettings{ FInfo: os.Stdout, @@ -141,3 +144,25 @@ func exchangeKeys(node1, node2 *sNode) (layer2.IParticipantKey, layer2.IParticip return pubKey1, pubKey2 } + +var ( + handler = func(ctx context.Context, n anonymity.INode, pKey layer2.IParticipantKey, b []byte) ([]byte, error) { + numBytes := [encoding.CSizeUint64]byte{} + copy(numBytes[:], b) + + num := encoding.BytesToUint64(numBytes) + msg := "ping" + if num%2 == 1 { + msg = "pong" + } + fmt.Printf("%s-%d\n", msg, num) + + numBytes = encoding.Uint64ToBytes(num + 1) + _ = n.SendPayload( + ctx, + pKey, + payload.NewPayload64(uint64(nodeRouter), numBytes[:]), + ) + return nil, nil + } +) diff --git a/pkg/anonymity/qb/examples/ping-pong/main.go b/pkg/anonymity/qb/examples/ping-pong/main.go index 455520a3..29eb1801 100644 --- a/pkg/anonymity/qb/examples/ping-pong/main.go +++ b/pkg/anonymity/qb/examples/ping-pong/main.go @@ -2,11 +2,8 @@ package main import ( "context" - "fmt" "time" - anonymity "github.com/number571/go-peer/pkg/anonymity/qb" - "github.com/number571/go-peer/pkg/crypto/scheme/layer2" "github.com/number571/go-peer/pkg/encoding" "github.com/number571/go-peer/pkg/payload" ) @@ -16,28 +13,6 @@ const ( nodeRouter = uint32(0xA557711A) ) -var ( - handler = func(ctx context.Context, n anonymity.INode, pKey layer2.IParticipantKey, b []byte) ([]byte, error) { - numBytes := [encoding.CSizeUint64]byte{} - copy(numBytes[:], b) - - num := encoding.BytesToUint64(numBytes) - msg := "ping" - if num%2 == 1 { - msg = "pong" - } - fmt.Printf("%s-%d\n", msg, num) - - numBytes = encoding.Uint64ToBytes(num + 1) - _ = n.SendPayload( - ctx, - pKey, - payload.NewPayload64(uint64(nodeRouter), numBytes[:]), - ) - return nil, nil - } -) - func init() { printTagVersion() } @@ -61,7 +36,6 @@ func main() { func runClientNode() *sNode { ctx := context.Background() node := newNode("cnode", "") - node.fAnonymity.HandleFunc(nodeRouter, handler) go func() { _ = node.fAnonymity.Run(ctx) }() _ = node.fNetwork.AddConnection(ctx, nodeAddress) @@ -72,7 +46,6 @@ func runClientNode() *sNode { func runServiceNode() *sNode { ctx := context.Background() node := newNode("snode", nodeAddress) - node.fAnonymity.HandleFunc(nodeRouter, handler) go func() { _ = node.fAnonymity.Run(ctx) }() go func() { _ = node.fNetwork.Run(ctx) }() diff --git a/pkg/anonymity/qb/types.go b/pkg/anonymity/qb/types.go index 0a25c452..64b32456 100644 --- a/pkg/anonymity/qb/types.go +++ b/pkg/anonymity/qb/types.go @@ -19,7 +19,6 @@ type ( type INode interface { types.IRunner - HandleFunc(uint32, IHandlerF) INode GetLogger() logger.ILogger GetSettings() ISettings diff --git a/test/result/badge_codelines.svg b/test/result/badge_codelines.svg index 4484af0e..d56e5c55 100644 --- a/test/result/badge_codelines.svg +++ b/test/result/badge_codelines.svg @@ -1 +1 @@ -code lines: 12931code lines12931 \ No newline at end of file +code lines: 12887code lines12887 \ No newline at end of file diff --git a/test/result/coverage.svg b/test/result/coverage.svg index 5060eeb6..53d67487 100644 --- a/test/result/coverage.svg +++ b/test/result/coverage.svg @@ -7,7 +7,7 @@ > - + - + - + - + anonymity/qb @@ -52,7 +52,7 @@ - + - + encoding @@ -78,12 +78,12 @@ - + logger @@ -91,12 +91,12 @@ - + network @@ -104,12 +104,12 @@ - + payload @@ -117,12 +117,12 @@ - + state @@ -130,12 +130,12 @@ - + storage @@ -143,12 +143,12 @@ - + types/converter.go @@ -156,12 +156,12 @@ - + action.go @@ -169,18 +169,18 @@ - + - + anonymity.go @@ -188,18 +188,18 @@ - + - + head.go @@ -207,12 +207,12 @@ - + logger/log_builder.go @@ -220,12 +220,12 @@ - + queue @@ -233,12 +233,12 @@ - + settings.go @@ -246,7 +246,7 @@ - + - + hashing @@ -272,18 +272,18 @@ - + - + puzzle/puzzle.go @@ -291,12 +291,12 @@ - + random/random.go @@ -304,7 +304,7 @@ - + - + symmetric @@ -330,12 +330,12 @@ - + bytes.go @@ -343,30 +343,30 @@ - + - + - + - + serialize_yaml.go @@ -374,12 +374,12 @@ - + logger.go @@ -387,18 +387,18 @@ - + - + conn @@ -406,12 +406,12 @@ - + connkeeper @@ -419,12 +419,12 @@ - + network.go @@ -432,12 +432,12 @@ - + settings.go @@ -445,12 +445,12 @@ - + joiner @@ -458,12 +458,12 @@ - + payload32.go @@ -471,12 +471,12 @@ - + payload64.go @@ -484,12 +484,12 @@ - + state.go @@ -497,12 +497,12 @@ - + cache/lru.go @@ -510,12 +510,12 @@ - + database @@ -523,18 +523,18 @@ - + - + queue.go @@ -542,12 +542,12 @@ - + settings.go @@ -555,12 +555,12 @@ - + dsa.go @@ -568,12 +568,12 @@ - + kem.go @@ -581,7 +581,7 @@ - + - + hashing.go @@ -607,12 +607,12 @@ - + hmac.go @@ -620,12 +620,12 @@ - + layer1 @@ -633,7 +633,7 @@ - + - + cfb.go @@ -659,12 +659,12 @@ - + cipher.go @@ -672,12 +672,12 @@ - + gcm.go @@ -685,12 +685,12 @@ - + conn.go @@ -698,12 +698,12 @@ - + settings.go @@ -711,12 +711,12 @@ - + connkeeper.go @@ -724,18 +724,18 @@ - + - + joiner32.go @@ -743,12 +743,12 @@ - + database.go @@ -756,12 +756,12 @@ - + message.go @@ -769,12 +769,12 @@ - + settings.go @@ -782,7 +782,7 @@ - + - + kcontainer.go @@ -808,12 +808,12 @@ - + symmetric @@ -821,12 +821,12 @@ - + message.go @@ -834,12 +834,12 @@ - + scheme.go @@ -847,12 +847,12 @@ - + scheme.go