update

CHANGELOG.md

@@ -12,7 +12,7 @@
 
 ### CHANGES
 
-- Update `pkg/types`: Run method of Runner now contains context.Context parameter
+- Update `pkg`: Change functions / methods to context.Context implementation
 
 ### BUG FIXES
 

TODO.md

@@ -9,8 +9,6 @@
 7. Append example of use functions (godoc)
 8. Append errors (as traffic/database) for all interface methods
 9. Append Batch's to database
-10. Think about append void bytes into network.IMessage
-11. Some methods to context.Context implementation
 
 ### Tests
 

cmd/hidden_lake/composite/service_messenger/cmd/hlc_sm/main.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"context"
+	"errors"
 	"os"
 	"os/signal"
 	"syscall"
@@ -25,7 +26,7 @@ func main() {
 
 	go func() {
 		defer func() { closed <- struct{}{} }()
-		if err := app.Run(ctx); err != nil {
+		if err := app.Run(ctx); err != nil && !errors.Is(err, context.Canceled) {
 			panic(err)
 		}
 	}()

cmd/hidden_lake/composite/service_traffic/cmd/hlc_st/main.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"context"
+	"errors"
 	"os"
 	"os/signal"
 	"syscall"
@@ -25,7 +26,7 @@ func main() {
 
 	go func() {
 		defer func() { closed <- struct{}{} }()
-		if err := app.Run(ctx); err != nil {
+		if err := app.Run(ctx); err != nil && !errors.Is(err, context.Canceled) {
 			panic(err)
 		}
 	}()

cmd/hidden_lake/composite/service_traffic_messenger/cmd/hlc_stm/main.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"context"
+	"errors"
 	"os"
 	"os/signal"
 	"syscall"
@@ -25,7 +26,7 @@ func main() {
 
 	go func() {
 		defer func() { closed <- struct{}{} }()
-		if err := app.Run(ctx); err != nil {
+		if err := app.Run(ctx); err != nil && !errors.Is(err, context.Canceled) {
 			panic(err)
 		}
 	}()

cmd/hidden_lake/loader/cmd/hll/main.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"context"
+	"errors"
 	"os"
 	"os/signal"
 	"syscall"
@@ -27,7 +28,7 @@ func main() {
 
 	go func() {
 		defer func() { closed <- struct{}{} }()
-		if err := app.Run(ctx); err != nil {
+		if err := app.Run(ctx); err != nil && !errors.Is(err, context.Canceled) {
 			panic(err)
 		}
 	}()

cmd/hidden_lake/loader/pkg/app/app.go

@@ -99,7 +99,7 @@ func (p *sApp) Run(pCtx context.Context) error {
 
 	select {
 	case <-pCtx.Done():
-		return nil
+		return pCtx.Err()
 	case err := <-chErr:
 		return fmt.Errorf("got run error: %w", err)
 	}

cmd/hidden_lake/messenger/cmd/hlm/main.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"context"
+	"errors"
 	"os"
 	"os/signal"
 	"syscall"
@@ -27,7 +28,7 @@ func main() {
 
 	go func() {
 		defer func() { closed <- struct{}{} }()
-		if err := app.Run(ctx); err != nil {
+		if err := app.Run(ctx); err != nil && !errors.Is(err, context.Canceled) {
 			panic(err)
 		}
 	}()

cmd/hidden_lake/messenger/pkg/app/app.go

@@ -109,7 +109,7 @@ func (p *sApp) Run(pCtx context.Context) error {
 
 	select {
 	case <-pCtx.Done():
-		return nil
+		return pCtx.Err()
 	case err := <-chErr:
 		return fmt.Errorf("got run error: %w", err)
 	}

cmd/hidden_lake/service/cmd/hls/main.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"context"
+	"errors"
 	"os"
 	"os/signal"
 	"syscall"
@@ -27,7 +28,7 @@ func main() {
 
 	go func() {
 		defer func() { closed <- struct{}{} }()
-		if err := app.Run(ctx); err != nil {
+		if err := app.Run(ctx); err != nil && !errors.Is(err, context.Canceled) {
 			panic(err)
 		}
 	}()

cmd/hidden_lake/service/internal/handler/config_connects.go

@@ -1,6 +1,7 @@
 package handler
 
 import (
+	"context"
 	"io"
 	"net/http"
 	"strings"
@@ -14,7 +15,7 @@ import (
 	"github.com/number571/go-peer/pkg/network/anonymity"
 )
 
-func HandleConfigConnectsAPI(pWrapper config.IWrapper, pLogger logger.ILogger, pNode anonymity.INode) http.HandlerFunc {
+func HandleConfigConnectsAPI(pCtx context.Context, pWrapper config.IWrapper, pLogger logger.ILogger, pNode anonymity.INode) http.HandlerFunc {
 	return func(pW http.ResponseWriter, pR *http.Request) {
 		logBuilder := http_logger.NewLogBuilder(pkg_settings.CServiceName, pR)
 
@@ -57,7 +58,7 @@ func HandleConfigConnectsAPI(pWrapper config.IWrapper, pLogger logger.ILogger, p
 				return
 			}
 
-			_ = pNode.GetNetworkNode().AddConnection(connect) // connection may be refused (closed)
+			_ = pNode.GetNetworkNode().AddConnection(pCtx, connect) // connection may be refused (closed)
 
 			pLogger.PushInfo(logBuilder.WithMessage(http_logger.CLogSuccess))
 			api.Response(pW, http.StatusOK, "success: update connections")

cmd/hidden_lake/service/internal/handler/config_connects_test.go

@@ -17,7 +17,7 @@ func TestHandleConnectsAPI(t *testing.T) {
 	pathCfg := fmt.Sprintf(tcPathConfigTemplate, 0)
 	pathDB := fmt.Sprintf(tcPathDBTemplate, 0)
 
-	wcfg, node, cancel, srv := testAllCreate(pathCfg, pathDB, testutils.TgAddrs[6])
+	wcfg, node, _, cancel, srv := testAllCreate(pathCfg, pathDB, testutils.TgAddrs[6])
 	defer testAllFree(node, cancel, srv, pathCfg, pathDB)
 
 	client := hls_client.NewClient(

cmd/hidden_lake/service/internal/handler/config_friends_test.go

@@ -18,7 +18,7 @@ func TestHandleFriendsAPI(t *testing.T) {
 	pathCfg := fmt.Sprintf(tcPathConfigTemplate, 1)
 	pathDB := fmt.Sprintf(tcPathDBTemplate, 1)
 
-	wcfg, node, cancel, srv := testAllCreate(pathCfg, pathDB, testutils.TgAddrs[7])
+	wcfg, node, _, cancel, srv := testAllCreate(pathCfg, pathDB, testutils.TgAddrs[7])
 	defer testAllFree(node, cancel, srv, pathCfg, pathDB)
 
 	client := hls_client.NewClient(

cmd/hidden_lake/service/internal/handler/config_settings_test.go

@@ -17,7 +17,7 @@ func TestHandleConfigSettingsAPI(t *testing.T) {
 	pathCfg := fmt.Sprintf(tcPathConfigTemplate, 2)
 	pathDB := fmt.Sprintf(tcPathDBTemplate, 2)
 
-	_, node, cancel, srv := testAllCreate(pathCfg, pathDB, addr)
+	_, node, _, cancel, srv := testAllCreate(pathCfg, pathDB, addr)
 	defer testAllFree(node, cancel, srv, pathCfg, pathDB)
 
 	client := hls_client.NewClient(

cmd/hidden_lake/service/internal/handler/handler_test.go

@@ -103,12 +103,12 @@ func testEchoPage(w http.ResponseWriter, r *http.Request) {
 	json.NewEncoder(w).Encode(resp)
 }
 
-func testAllCreate(cfgPath, dbPath, srvAddr string) (config.IWrapper, anonymity.INode, context.CancelFunc, *http.Server) {
+func testAllCreate(cfgPath, dbPath, srvAddr string) (config.IWrapper, anonymity.INode, context.Context, context.CancelFunc, *http.Server) {
 	wcfg := testNewWrapper(cfgPath)
-	node, cancel := testRunNewNode(dbPath, "")
-	srvc := testRunService(wcfg, node, srvAddr)
+	node, ctx, cancel := testRunNewNode(dbPath, "")
+	srvc := testRunService(ctx, wcfg, node, srvAddr)
 	time.Sleep(200 * time.Millisecond)
-	return wcfg, node, cancel, srvc
+	return wcfg, node, ctx, cancel, srvc
 }
 
 func testAllFree(node anonymity.INode, cancel context.CancelFunc, srv *http.Server, pathCfg, pathDB string) {
@@ -124,7 +124,7 @@ func testAllFree(node anonymity.INode, cancel context.CancelFunc, srv *http.Serv
 	})
 }
 
-func testRunService(wcfg config.IWrapper, node anonymity.INode, addr string) *http.Server {
+func testRunService(ctx context.Context, wcfg config.IWrapper, node anonymity.INode, addr string) *http.Server {
 	mux := http.NewServeMux()
 
 	logger := logger.NewLogger(
@@ -134,10 +134,10 @@ func testRunService(wcfg config.IWrapper, node anonymity.INode, addr string) *ht
 
 	mux.HandleFunc(pkg_settings.CHandleIndexPath, HandleIndexAPI(logger))
 	mux.HandleFunc(pkg_settings.CHandleConfigSettingsPath, HandleConfigSettingsAPI(wcfg, logger))
-	mux.HandleFunc(pkg_settings.CHandleConfigConnectsPath, HandleConfigConnectsAPI(wcfg, logger, node))
+	mux.HandleFunc(pkg_settings.CHandleConfigConnectsPath, HandleConfigConnectsAPI(ctx, wcfg, logger, node))
 	mux.HandleFunc(pkg_settings.CHandleConfigFriendsPath, HandleConfigFriendsAPI(wcfg, logger, node))
 	mux.HandleFunc(pkg_settings.CHandleNetworkOnlinePath, HandleNetworkOnlineAPI(logger, node))
-	mux.HandleFunc(pkg_settings.CHandleNetworkRequestPath, HandleNetworkRequestAPI(wcfg, logger, node))
+	mux.HandleFunc(pkg_settings.CHandleNetworkRequestPath, HandleNetworkRequestAPI(ctx, wcfg, logger, node))
 	mux.HandleFunc(pkg_settings.CHandleNetworkKeyPath, HandleNetworkKeyAPI(wcfg, logger, node))
 	mux.HandleFunc(pkg_settings.CHandleNodeKeyPath, HandleNodeKeyAPI(wcfg, logger, node))
 
@@ -162,11 +162,11 @@ func testNewWrapper(cfgPath string) config.IWrapper {
 	return config.NewWrapper(cfg)
 }
 
-func testRunNewNode(dbPath, addr string) (anonymity.INode, context.CancelFunc) {
+func testRunNewNode(dbPath, addr string) (anonymity.INode, context.Context, context.CancelFunc) {
 	node := testNewNode(dbPath, addr).HandleFunc(pkg_settings.CServiceMask, nil)
 	ctx, cancel := context.WithCancel(context.Background())
 	go func() { _ = node.Run(ctx) }()
-	return node, cancel
+	return node, ctx, cancel
 }
 
 func testNewNode(dbPath, addr string) anonymity.INode {

cmd/hidden_lake/service/internal/handler/index_test.go

@@ -18,7 +18,7 @@ func TestHandleIndexAPI(t *testing.T) {
 	pathCfg := fmt.Sprintf(tcPathConfigTemplate, 3)
 	pathDB := fmt.Sprintf(tcPathDBTemplate, 3)
 
-	_, node, cancel, srv := testAllCreate(pathCfg, pathDB, addr)
+	_, node, _, cancel, srv := testAllCreate(pathCfg, pathDB, addr)
 	defer testAllFree(node, cancel, srv, pathCfg, pathDB)
 
 	client := hls_client.NewClient(

cmd/hidden_lake/service/internal/handler/network_key_test.go

@@ -17,7 +17,7 @@ func TestHandleNetworkKeyAPI(t *testing.T) {
 	pathCfg := fmt.Sprintf(tcPathConfigTemplate, 4)
 	pathDB := fmt.Sprintf(tcPathDBTemplate, 4)
 
-	wcfg, node, cancel, srv := testAllCreate(pathCfg, pathDB, testutils.TgAddrs[25])
+	wcfg, node, _, cancel, srv := testAllCreate(pathCfg, pathDB, testutils.TgAddrs[25])
 	defer testAllFree(node, cancel, srv, pathCfg, pathDB)
 
 	client := hls_client.NewClient(

cmd/hidden_lake/service/internal/handler/network_online_test.go

@@ -25,7 +25,7 @@ func TestHandleOnlineAPI(t *testing.T) {
 	pathCfg := fmt.Sprintf(tcPathConfigTemplate, 6)
 	pathDB := fmt.Sprintf(tcPathDBTemplate, 6)
 
-	_, node, cancel, srv := testAllCreate(pathCfg, pathDB, testutils.TgAddrs[12])
+	_, node, ctx, cancel, srv := testAllCreate(pathCfg, pathDB, testutils.TgAddrs[12])
 	defer testAllFree(node, cancel, srv, pathCfg, pathDB)
 
 	pushNode, pushCancel := testAllOnlineCreate(pathCfg, pathDB)
@@ -39,7 +39,7 @@ func TestHandleOnlineAPI(t *testing.T) {
 		),
 	)
 
-	node.GetNetworkNode().AddConnection(testutils.TgAddrs[13])
+	node.GetNetworkNode().AddConnection(ctx, testutils.TgAddrs[13])
 	node.GetListPubKeys().AddPubKey(asymmetric.LoadRSAPrivKey(testutils.Tc1PrivKey1024).GetPubKey())
 
 	testGetOnlines(t, client, node)
@@ -109,7 +109,7 @@ func testAllOnlineFree(node anonymity.INode, cancel context.CancelFunc, pathCfg,
 }
 
 func testOnlinePushNode(cfgPath, dbPath string) (anonymity.INode, context.CancelFunc) {
-	node, cancel := testRunNewNode(dbPath, testutils.TgAddrs[13])
+	node, ctx, cancel := testRunNewNode(dbPath, testutils.TgAddrs[13])
 
 	cfg, err := config.BuildConfig(cfgPath, &config.SConfig{
 		FSettings: &config.SConfigSettings{
@@ -135,9 +135,7 @@ func testOnlinePushNode(cfgPath, dbPath string) (anonymity.INode, context.Cancel
 	)
 	node.GetListPubKeys().AddPubKey(asymmetric.LoadRSAPrivKey(testutils.Tc1PrivKey1024).GetPubKey())
 
-	if err := node.GetNetworkNode().Listen(); err != nil {
-		return nil, cancel
-	}
+	go func() { _ = node.GetNetworkNode().Listen(ctx) }()
 
 	return node, cancel
 }

cmd/hidden_lake/service/internal/handler/network_request.go

@@ -1,6 +1,7 @@
 package handler
 
 import (
+	"context"
 	"encoding/json"
 	"net/http"
 
@@ -23,7 +24,7 @@ const (
 	cErrorLoadRequest
 )
 
-func HandleNetworkRequestAPI(pWrapper config.IWrapper, pLogger logger.ILogger, pNode anonymity.INode) http.HandlerFunc {
+func HandleNetworkRequestAPI(pCtx context.Context, pWrapper config.IWrapper, pLogger logger.ILogger, pNode anonymity.INode) http.HandlerFunc {
 	return func(pW http.ResponseWriter, pR *http.Request) {
 		logBuilder := http_logger.NewLogBuilder(pkg_settings.CServiceName, pR)
 
@@ -64,6 +65,7 @@ func HandleNetworkRequestAPI(pWrapper config.IWrapper, pLogger logger.ILogger, p
 		switch pR.Method {
 		case http.MethodPut:
 			err := pNode.BroadcastPayload(
+				pCtx,
 				pubKey,
 				adapters.NewPayload(pkg_settings.CServiceMask, data),
 			)
@@ -80,6 +82,7 @@ func HandleNetworkRequestAPI(pWrapper config.IWrapper, pLogger logger.ILogger, p
 
 		case http.MethodPost:
 			respBytes, err := pNode.FetchPayload(
+				pCtx,
 				pubKey,
 				adapters.NewPayload(pkg_settings.CServiceMask, data),
 			)

cmd/hidden_lake/service/internal/handler/network_request_test.go

@@ -26,7 +26,7 @@ func TestHandleRequestAPI(t *testing.T) {
 	pathCfg := fmt.Sprintf(tcPathConfigTemplate, 7)
 	pathDB := fmt.Sprintf(tcPathDBTemplate, 7)
 
-	_, node, cancel, srv := testAllCreate(pathCfg, pathDB, testutils.TgAddrs[9])
+	_, node, ctx, cancel, srv := testAllCreate(pathCfg, pathDB, testutils.TgAddrs[9])
 	defer testAllFree(node, cancel, srv, pathCfg, pathDB)
 
 	pushNode, pushCancel, pushSrv := testAllPushCreate(pathCfg, pathDB)
@@ -40,7 +40,7 @@ func TestHandleRequestAPI(t *testing.T) {
 		),
 	)
 
-	node.GetNetworkNode().AddConnection(testutils.TgAddrs[11])
+	node.GetNetworkNode().AddConnection(ctx, testutils.TgAddrs[11])
 	node.GetListPubKeys().AddPubKey(asymmetric.LoadRSAPrivKey(testutils.Tc1PrivKey1024).GetPubKey())
 
 	testBroadcast(t, client)
@@ -111,7 +111,7 @@ func testAllPushFree(node anonymity.INode, cancel context.CancelFunc, srv *http.
 }
 
 func testNewPushNode(cfgPath, dbPath string) (anonymity.INode, context.CancelFunc) {
-	node, cancel := testRunNewNode(dbPath, testutils.TgAddrs[11])
+	node, ctx, cancel := testRunNewNode(dbPath, testutils.TgAddrs[11])
 	rawCFG := &config.SConfig{
 		FSettings: &config.SConfigSettings{
 			FMessageSizeBytes: testutils.TCMessageSize,
@@ -141,9 +141,7 @@ func testNewPushNode(cfgPath, dbPath string) (anonymity.INode, context.CancelFun
 	)
 	node.GetListPubKeys().AddPubKey(asymmetric.LoadRSAPrivKey(testutils.Tc1PrivKey1024).GetPubKey())
 
-	if err := node.GetNetworkNode().Listen(); err != nil {
-		return nil, cancel
-	}
+	go func() { _ = node.GetNetworkNode().Listen(ctx) }()
 
 	return node, cancel
 }

cmd/hidden_lake/service/internal/handler/node_key_test.go

@@ -16,7 +16,7 @@ func TestHandlePubKeyAPI(t *testing.T) {
 	pathCfg := fmt.Sprintf(tcPathConfigTemplate, 8)
 	pathDB := fmt.Sprintf(tcPathDBTemplate, 8)
 
-	_, node, cancel, srv := testAllCreate(pathCfg, pathDB, testutils.TgAddrs[8])
+	_, node, _, cancel, srv := testAllCreate(pathCfg, pathDB, testutils.TgAddrs[8])
 	defer testAllFree(node, cancel, srv, pathCfg, pathDB)
 
 	client := hls_client.NewClient(

cmd/hidden_lake/service/internal/handler/service_tcp.go

@@ -2,6 +2,7 @@ package handler
 
 import (
 	"bytes"
+	"context"
 	"fmt"
 	"io"
 	"net/http"
@@ -20,7 +21,7 @@ import (
 )
 
 func HandleServiceTCP(pCfg config.IConfig, pLogger logger.ILogger) anonymity.IHandlerF {
-	return func(_ anonymity.INode, sender asymmetric.IPubKey, reqBytes []byte) ([]byte, error) {
+	return func(pCtx context.Context, _ anonymity.INode, sender asymmetric.IPubKey, reqBytes []byte) ([]byte, error) {
 		logBuilder := anon_logger.NewLogBuilder(pkg_settings.CServiceName)
 
 		// enrich logger

cmd/hidden_lake/service/internal/handler/service_tcp_test.go

@@ -90,7 +90,7 @@ func testStartNodeHLS(t *testing.T) (anonymity.INode, context.CancelFunc, error)
 		return nil, nil, err
 	}
 
-	node, cancel := testRunNewNode(fmt.Sprintf(tcPathDBTemplate, 9), testutils.TgAddrs[4])
+	node, ctx, cancel := testRunNewNode(fmt.Sprintf(tcPathDBTemplate, 9), testutils.TgAddrs[4])
 	if node == nil {
 		return nil, nil, fmt.Errorf("node is not running")
 	}
@@ -107,10 +107,10 @@ func testStartNodeHLS(t *testing.T) (anonymity.INode, context.CancelFunc, error)
 	)
 	node.GetListPubKeys().AddPubKey(asymmetric.LoadRSAPrivKey(testutils.Tc1PrivKey1024).GetPubKey())
 
-	if err := node.GetNetworkNode().Listen(); err != nil {
-		t.Error(err)
-		return nil, cancel, nil
-	}
+	go func() {
+		_ = node.GetNetworkNode().Listen(ctx)
+	}()
+
 	return node, cancel, nil
 }
 
@@ -119,13 +119,13 @@ func testStartNodeHLS(t *testing.T) (anonymity.INode, context.CancelFunc, error)
 func testStartClientHLS() (anonymity.INode, context.CancelFunc, error) {
 	time.Sleep(time.Second)
 
-	node, cancel := testRunNewNode(fmt.Sprintf(tcPathDBTemplate, 10), "")
+	node, ctx, cancel := testRunNewNode(fmt.Sprintf(tcPathDBTemplate, 10), "")
 	if node == nil {
 		return nil, cancel, fmt.Errorf("node is not running")
 	}
 	node.GetListPubKeys().AddPubKey(asymmetric.LoadRSAPrivKey(testutils.Tc1PrivKey1024).GetPubKey())
 
-	if err := node.GetNetworkNode().AddConnection(testutils.TgAddrs[4]); err != nil {
+	if err := node.GetNetworkNode().AddConnection(ctx, testutils.TgAddrs[4]); err != nil {
 		return nil, cancel, err
 	}
 
@@ -140,7 +140,7 @@ func testStartClientHLS() (anonymity.INode, context.CancelFunc, error) {
 	)
 
 	pubKey := asymmetric.LoadRSAPrivKey(testutils.Tc1PrivKey1024).GetPubKey()
-	respBytes, err := node.FetchPayload(pubKey, pld)
+	respBytes, err := node.FetchPayload(ctx, pubKey, pld)
 	if err != nil {
 		return node, cancel, err
 	}

cmd/hidden_lake/service/pkg/app/app.go

@@ -91,7 +91,7 @@ func (p *sApp) Run(pCtx context.Context) error {
 		_ = p.fState.Disable(disableFunc)
 	}()
 
-	p.initServiceHTTP()
+	p.initServiceHTTP(pCtx)
 	p.initServicePPROF()
 
 	chErr := make(chan error)
@@ -128,7 +128,7 @@ func (p *sApp) Run(pCtx context.Context) error {
 		}
 
 		// run node in server mode
-		err := p.fNode.GetNetworkNode().Listen()
+		err := p.fNode.GetNetworkNode().Listen(pCtx)
 		if err != nil && !errors.Is(err, net.ErrClosed) {
 			chErr <- err
 			return
@@ -151,7 +151,7 @@ func (p *sApp) Run(pCtx context.Context) error {
 
 	select {
 	case <-pCtx.Done():
-		return nil
+		return pCtx.Err()
 	case err := <-chErr:
 		return fmt.Errorf("got run error: %w", err)
 	}

cmd/hidden_lake/service/pkg/app/app_test.go

@@ -2,6 +2,7 @@ package app
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"net/http"
 	"os"
@@ -57,7 +58,7 @@ func TestApp(t *testing.T) {
 	defer cancel()
 
 	go func() {
-		if err := app.Run(ctx); err != nil {
+		if err := app.Run(ctx); err != nil && !errors.Is(err, context.Canceled) {
 			t.Error(err)
 			return
 		}
@@ -100,7 +101,7 @@ func TestApp(t *testing.T) {
 
 	// try twice running
 	go func() {
-		if err := app.Run(ctx1); err != nil {
+		if err := app.Run(ctx1); err != nil && !errors.Is(err, context.Canceled) {
 			t.Error(err)
 			return
 		}

cmd/hidden_lake/service/pkg/app/init_service.go

@@ -1,21 +1,22 @@
 package app
 
 import (
+	"context"
 	"net/http"
 
 	"github.com/number571/go-peer/cmd/hidden_lake/service/internal/handler"
 	pkg_settings "github.com/number571/go-peer/cmd/hidden_lake/service/pkg/settings"
 )
 
-func (p *sApp) initServiceHTTP() {
+func (p *sApp) initServiceHTTP(pCtx context.Context) {
 	mux := http.NewServeMux()
 
 	mux.HandleFunc(pkg_settings.CHandleIndexPath, handler.HandleIndexAPI(p.fHTTPLogger))
 	mux.HandleFunc(pkg_settings.CHandleConfigSettingsPath, handler.HandleConfigSettingsAPI(p.fWrapper, p.fHTTPLogger))
-	mux.HandleFunc(pkg_settings.CHandleConfigConnectsPath, handler.HandleConfigConnectsAPI(p.fWrapper, p.fHTTPLogger, p.fNode))
+	mux.HandleFunc(pkg_settings.CHandleConfigConnectsPath, handler.HandleConfigConnectsAPI(pCtx, p.fWrapper, p.fHTTPLogger, p.fNode))
 	mux.HandleFunc(pkg_settings.CHandleConfigFriendsPath, handler.HandleConfigFriendsAPI(p.fWrapper, p.fHTTPLogger, p.fNode))
 	mux.HandleFunc(pkg_settings.CHandleNetworkOnlinePath, handler.HandleNetworkOnlineAPI(p.fHTTPLogger, p.fNode))
-	mux.HandleFunc(pkg_settings.CHandleNetworkRequestPath, handler.HandleNetworkRequestAPI(p.fWrapper, p.fHTTPLogger, p.fNode))
+	mux.HandleFunc(pkg_settings.CHandleNetworkRequestPath, handler.HandleNetworkRequestAPI(pCtx, p.fWrapper, p.fHTTPLogger, p.fNode))
 	mux.HandleFunc(pkg_settings.CHandleNetworkKeyPath, handler.HandleNetworkKeyAPI(p.fWrapper, p.fHTTPLogger, p.fNode))
 	mux.HandleFunc(pkg_settings.CHandleNodeKeyPath, handler.HandleNodeKeyAPI(p.fWrapper, p.fHTTPLogger, p.fNode))
 

cmd/hidden_lake/traffic/cmd/hlt/main.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"context"
+	"errors"
 	"os"
 	"os/signal"
 	"syscall"
@@ -27,7 +28,7 @@ func main() {
 
 	go func() {
 		defer func() { closed <- struct{}{} }()
-		if err := app.Run(ctx); err != nil {
+		if err := app.Run(ctx); err != nil && !errors.Is(err, context.Canceled) {
 			panic(err)
 		}
 	}()

cmd/hidden_lake/traffic/internal/handler/handler_test.go

@@ -95,7 +95,7 @@ func testRunService(wDB database.IWrapperDB, addr string, addrNode string) (*htt
 		conn_keeper.NewSettings(connKeeperSettings),
 		testNewNetworkNode("").HandleFunc(
 			1, // default value
-			func(_ network.INode, _ conn.IConn, _ net_message.IMessage) error {
+			func(_ context.Context, _ network.INode, _ conn.IConn, _ net_message.IMessage) error {
 				// pass response actions
 				return nil
 			},
@@ -143,7 +143,7 @@ func testRunService(wDB database.IWrapperDB, addr string, addrNode string) (*htt
 
 	mux.HandleFunc(pkg_settings.CHandleIndexPath, HandleIndexAPI(logger))
 	mux.HandleFunc(pkg_settings.CHandleHashesPath, HandleHashesAPI(wDB, logger))
-	mux.HandleFunc(pkg_settings.CHandleMessagePath, HandleMessageAPI(cfg, wDB, logger, logger, node))
+	mux.HandleFunc(pkg_settings.CHandleMessagePath, HandleMessageAPI(ctx, cfg, wDB, logger, logger, node))
 
 	srv := &http.Server{
 		Addr:    addr,

cmd/hidden_lake/traffic/internal/handler/message.go

@@ -1,6 +1,7 @@
 package handler
 
 import (
+	"context"
 	"io"
 	"net/http"
 
@@ -16,7 +17,7 @@ import (
 	net_message "github.com/number571/go-peer/pkg/network/message"
 )
 
-func HandleMessageAPI(pCfg config.IConfig, pWrapperDB database.IWrapperDB, pHTTPLogger, pAnonLogger logger.ILogger, pNode network.INode) http.HandlerFunc {
+func HandleMessageAPI(pCtx context.Context, pCfg config.IConfig, pWrapperDB database.IWrapperDB, pHTTPLogger, pAnonLogger logger.ILogger, pNode network.INode) http.HandlerFunc {
 	tcpHandler := HandleServiceTCP(pCfg, pWrapperDB, pAnonLogger)
 
 	return func(pW http.ResponseWriter, pR *http.Request) {
@@ -81,7 +82,7 @@ func HandleMessageAPI(pCfg config.IConfig, pWrapperDB database.IWrapperDB, pHTTP
 				return
 			}
 
-			if err := tcpHandler(pNode, nil, netMsg); err != nil {
+			if err := tcpHandler(pCtx, pNode, nil, netMsg); err != nil {
 				// internal logger
 				api.Response(pW, http.StatusBadRequest, "failed: handle message")
 				return

cmd/hidden_lake/traffic/internal/handler/service_tcp.go

@@ -1,6 +1,7 @@
 package handler
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"net/http"
@@ -22,7 +23,7 @@ import (
 func HandleServiceTCP(pCfg config.IConfig, pWrapperDB database.IWrapperDB, pLogger logger.ILogger) network.IHandlerF {
 	httpClient := &http.Client{Timeout: time.Minute}
 
-	return func(pNode network.INode, pConn conn.IConn, pNetMsg net_message.IMessage) error {
+	return func(pCtx context.Context, pNode network.INode, pConn conn.IConn, pNetMsg net_message.IMessage) error {
 		logBuilder := anon_logger.NewLogBuilder(hlt_settings.CServiceName)
 
 		// enrich logger
@@ -55,7 +56,7 @@ func HandleServiceTCP(pCfg config.IConfig, pWrapperDB database.IWrapperDB, pLogg
 			return fmt.Errorf("put message to database: %w", err)
 		}
 
-		if err := pNode.BroadcastMessage(pNetMsg); err != nil {
+		if err := pNode.BroadcastMessage(pCtx, pNetMsg); err != nil {
 			pLogger.PushWarn(logBuilder.WithType(anon_logger.CLogBaseBroadcast))
 			// need pass error (some of connections may be closed)
 		}

cmd/hidden_lake/traffic/pkg/app/app.go

@@ -98,7 +98,7 @@ func (p *sApp) Run(pCtx context.Context) error {
 		_ = p.fState.Disable(disableFunc)
 	}()
 
-	p.initServiceHTTP()
+	p.initServiceHTTP(pCtx)
 	p.initServicePPROF()
 
 	chErr := make(chan error)
@@ -141,7 +141,7 @@ func (p *sApp) Run(pCtx context.Context) error {
 			return
 		}
 
-		if err := p.fNode.Listen(); err != nil {
+		if err := p.fNode.Listen(pCtx); err != nil {
 			chErr <- err
 			return
 		}
@@ -149,7 +149,7 @@ func (p *sApp) Run(pCtx context.Context) error {
 
 	select {
 	case <-pCtx.Done():
-		return nil
+		return pCtx.Err()
 	case err := <-chErr:
 		return fmt.Errorf("got run error: %w", err)
 	}

cmd/hidden_lake/traffic/pkg/app/app_test.go

@@ -2,6 +2,7 @@ package app
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"net/http"
 	"os"
@@ -60,7 +61,7 @@ func TestApp(t *testing.T) {
 	app := NewApp(cfg, ".")
 
 	go func() {
-		if err := app.Run(ctx); err != nil {
+		if err := app.Run(ctx); err != nil && !errors.Is(err, context.Canceled) {
 			t.Error(err)
 			return
 		}
@@ -107,7 +108,7 @@ func TestApp(t *testing.T) {
 
 	// try twice running
 	go func() {
-		if err := app.Run(ctx1); err != nil {
+		if err := app.Run(ctx1); err != nil && !errors.Is(err, context.Canceled) {
 			t.Error(err)
 			return
 		}

cmd/hidden_lake/traffic/pkg/app/init_service.go

@@ -1,18 +1,19 @@
 package app
 
 import (
+	"context"
 	"net/http"
 
 	"github.com/number571/go-peer/cmd/hidden_lake/traffic/internal/handler"
 	pkg_settings "github.com/number571/go-peer/cmd/hidden_lake/traffic/pkg/settings"
 )
 
-func (p *sApp) initServiceHTTP() {
+func (p *sApp) initServiceHTTP(pCtx context.Context) {
 	mux := http.NewServeMux()
 
 	mux.HandleFunc(pkg_settings.CHandleIndexPath, handler.HandleIndexAPI(p.fHTTPLogger))
 	mux.HandleFunc(pkg_settings.CHandleHashesPath, handler.HandleHashesAPI(p.fWrapperDB, p.fHTTPLogger))
-	mux.HandleFunc(pkg_settings.CHandleMessagePath, handler.HandleMessageAPI(p.fConfig, p.fWrapperDB, p.fHTTPLogger, p.fAnonLogger, p.fNode))
+	mux.HandleFunc(pkg_settings.CHandleMessagePath, handler.HandleMessageAPI(pCtx, p.fConfig, p.fWrapperDB, p.fHTTPLogger, p.fAnonLogger, p.fNode))
 
 	p.fServiceHTTP = &http.Server{
 		Addr:    p.fConfig.GetAddress().GetHTTP(),

pkg/network/anonymity/anonymity.go

@@ -84,12 +84,7 @@ func (p *sNode) Run(pCtx context.Context) error {
 	}()
 
 	chErr := make(chan error)
-	go func() {
-		if err := p.fQueue.Run(pCtx); err != nil {
-			chErr <- fmt.Errorf("run queue: %w", err)
-		}
-		chErr <- nil
-	}()
+	go func() { chErr <- p.fQueue.Run(pCtx) }()
 
 	for {
 		select {
@@ -102,7 +97,7 @@ func (p *sNode) Run(pCtx context.Context) error {
 			}
 
 			logBuilder := anon_logger.NewLogBuilder(p.fSettings.GetServiceName())
-			if ok, _ := p.storeHashWithBroadcast(logBuilder, msg); !ok {
+			if ok, _ := p.storeHashWithBroadcast(pCtx, logBuilder, msg); !ok {
 				// internal logger
 				break
 			}
@@ -148,9 +143,9 @@ func (p *sNode) HandleFunc(pHead uint32, pHandle IHandlerF) INode {
 }
 
 // Send message without response waiting.
-func (p *sNode) BroadcastPayload(pRecv asymmetric.IPubKey, pPld adapters.IPayload) error {
+func (p *sNode) BroadcastPayload(pCtx context.Context, pRecv asymmetric.IPubKey, pPld adapters.IPayload) error {
 	// internal logger
-	if err := p.enqueuePayload(cIsRequest, pRecv, pPld.ToOrigin()); err != nil {
+	if err := p.enqueuePayload(pCtx, cIsRequest, pRecv, pPld.ToOrigin()); err != nil {
 		return fmt.Errorf("broadcast payload: %w", err)
 	}
 	return nil
@@ -158,7 +153,7 @@ func (p *sNode) BroadcastPayload(pRecv asymmetric.IPubKey, pPld adapters.IPayloa
 
 // Send message with response waiting.
 // Payload head must be uint32.
-func (p *sNode) FetchPayload(pRecv asymmetric.IPubKey, pPld adapters.IPayload) ([]byte, error) {
+func (p *sNode) FetchPayload(pCtx context.Context, pRecv asymmetric.IPubKey, pPld adapters.IPayload) ([]byte, error) {
 	headAction := uint32(random.NewStdPRNG().GetUint64())
 	newPld := payload.NewPayload(
 		joinHead(headAction, pPld.GetHead()).uint64(),
@@ -171,7 +166,7 @@ func (p *sNode) FetchPayload(pRecv asymmetric.IPubKey, pPld adapters.IPayload) (
 	defer p.delAction(actionKey)
 
 	// internal logger
-	if err := p.enqueuePayload(cIsRequest, pRecv, newPld); err != nil {
+	if err := p.enqueuePayload(pCtx, cIsRequest, pRecv, newPld); err != nil {
 		return nil, fmt.Errorf("fetch payload: %w", err)
 	}
 
@@ -183,13 +178,17 @@ func (p *sNode) FetchPayload(pRecv asymmetric.IPubKey, pPld adapters.IPayload) (
 	return resp, nil
 }
 
-func (p *sNode) send(pMsg message.IMessage) error {
+func (p *sNode) send(pCtx context.Context, pMsg message.IMessage) error {
 	for i := uint64(0); i <= p.fSettings.GetRetryEnqueue(); i++ {
-		if err := p.fQueue.EnqueueMessage(pMsg); err != nil {
-			time.Sleep(p.fQueue.GetSettings().GetDuration())
-			continue
+		if err := p.fQueue.EnqueueMessage(pMsg); err == nil {
+			return nil
+		}
+		select {
+		case <-pCtx.Done():
+			return pCtx.Err()
+		case <-time.After(p.fQueue.GetSettings().GetDuration()):
+			// next iter
 		}
-		return nil
 	}
 	return errors.New("enqueue message as send")
 }
@@ -211,14 +210,14 @@ func (p *sNode) recv(pActionKey string) ([]byte, error) {
 }
 
 func (p *sNode) handleWrapper() network.IHandlerF {
-	return func(_ network.INode, pConn conn.IConn, pMsg net_message.IMessage) error {
+	return func(pCtx context.Context, _ network.INode, pConn conn.IConn, pMsg net_message.IMessage) error {
 		logBuilder := anon_logger.NewLogBuilder(p.fSettings.GetServiceName())
 
 		// enrich logger
 		logBuilder.WithConn(pConn)
 
 		// try store hash of message
-		if ok, err := p.storeHashWithBroadcast(logBuilder, pMsg); !ok {
+		if ok, err := p.storeHashWithBroadcast(pCtx, logBuilder, pMsg); !ok {
 			// internal logger
 			if err != nil {
 				return fmt.Errorf("store hash with broadcast: %w", err)
@@ -278,7 +277,7 @@ func (p *sNode) handleWrapper() network.IHandlerF {
 			}
 
 			// response can be nil
-			resp, err := f(p, sender, unwrapBytes(body))
+			resp, err := f(pCtx, p, sender, unwrapBytes(body))
 			if err != nil {
 				p.fLogger.PushWarn(logBuilder.WithType(anon_logger.CLogWarnIncorrectResponse))
 				return nil
@@ -289,8 +288,8 @@ func (p *sNode) handleWrapper() network.IHandlerF {
 			}
 
 			// create the message and put this to the queue
+			_ = p.enqueuePayload(pCtx, cIsResponse, sender, payload.NewPayload(pld.GetHead(), resp))
 			// internal logger
-			_ = p.enqueuePayload(cIsResponse, sender, payload.NewPayload(pld.GetHead(), resp))
 			return nil
 
 		// undefined type of message (not request/response)
@@ -301,7 +300,7 @@ func (p *sNode) handleWrapper() network.IHandlerF {
 	}
 }
 
-func (p *sNode) enqueuePayload(pType iDataType, pRecv asymmetric.IPubKey, pPld payload.IPayload) error {
+func (p *sNode) enqueuePayload(pCtx context.Context, pType iDataType, pRecv asymmetric.IPubKey, pPld payload.IPayload) error {
 	logBuilder := anon_logger.NewLogBuilder(p.fSettings.GetServiceName())
 
 	// enrich logger
@@ -347,7 +346,7 @@ func (p *sNode) enqueuePayload(pType iDataType, pRecv asymmetric.IPubKey, pPld p
 		WithSize(size).
 		WithType(logType)
 
-	if err := p.send(msg); err != nil {
+	if err := p.send(pCtx, msg); err != nil {
 		p.fLogger.PushErro(logBuilder)
 		return fmt.Errorf("send message: %w", err)
 	}
@@ -356,7 +355,7 @@ func (p *sNode) enqueuePayload(pType iDataType, pRecv asymmetric.IPubKey, pPld p
 	return nil
 }
 
-func (p *sNode) storeHashWithBroadcast(pLogBuilder anon_logger.ILogBuilder, pNetMsg net_message.IMessage) (bool, error) {
+func (p *sNode) storeHashWithBroadcast(pCtx context.Context, pLogBuilder anon_logger.ILogBuilder, pNetMsg net_message.IMessage) (bool, error) {
 	var (
 		size      = len(pNetMsg.GetPayload().GetBody())
 		hash      = pNetMsg.GetHash()
@@ -400,7 +399,7 @@ func (p *sNode) storeHashWithBroadcast(pLogBuilder anon_logger.ILogBuilder, pNet
 	// do not send data if than already received
 	if !hashIsExist {
 		// broadcast message to network
-		if err := p.networkBroadcast(pNetMsg); err != nil {
+		if err := p.networkBroadcast(pCtx, pNetMsg); err != nil {
 			p.fLogger.PushWarn(pLogBuilder.WithType(anon_logger.CLogBaseBroadcast))
 			// need pass error (some of connections may be closed)
 		}
@@ -409,9 +408,9 @@ func (p *sNode) storeHashWithBroadcast(pLogBuilder anon_logger.ILogBuilder, pNet
 	return true, nil
 }
 
-func (p *sNode) networkBroadcast(pMsg net_message.IMessage) error {
+func (p *sNode) networkBroadcast(pCtx context.Context, pMsg net_message.IMessage) error {
 	// redirect message to another nodes
-	if err := p.fNetwork.BroadcastMessage(pMsg); err != nil {
+	if err := p.fNetwork.BroadcastMessage(pCtx, pMsg); err != nil {
 		return fmt.Errorf("network broadcast message: %w", err)
 	}
 	return nil

pkg/network/anonymity/anonymity_test.go

@@ -97,6 +97,8 @@ func TestComplexFetchPayload(t *testing.T) {
 	wg := sync.WaitGroup{}
 	wg.Add(tcIter)
 
+	ctx := context.Background()
+
 	for i := 0; i < tcIter; i++ {
 		go func(i int) {
 			defer wg.Done()
@@ -104,6 +106,7 @@ func TestComplexFetchPayload(t *testing.T) {
 
 			// nodes[1] -> nodes[0] -> nodes[2]
 			resp, err := nodes[0].FetchPayload(
+				ctx,
 				nodes[1].GetMessageQueue().GetClient().GetPubKey(),
 				adapters.NewPayload(testutils.TcHead, []byte(reqBody)),
 			)
@@ -137,8 +140,11 @@ func TestF2FWithoutFriends(t *testing.T) {
 	nodes[0].GetListPubKeys().DelPubKey(nodes[1].GetMessageQueue().GetClient().GetPubKey())
 	nodes[1].GetListPubKeys().DelPubKey(nodes[0].GetMessageQueue().GetClient().GetPubKey())
 
+	ctx := context.Background()
+
 	// nodes[1] -> nodes[0] -> nodes[2]
 	_, err := nodes[0].FetchPayload(
+		ctx,
 		nodes[1].GetMessageQueue().GetClient().GetPubKey(),
 		adapters.NewPayload(testutils.TcHead, []byte(testutils.TcBody)),
 	)
@@ -196,13 +202,16 @@ func TestFetchPayload(t *testing.T) {
 
 	nodes[1].HandleFunc(
 		testutils.TcHead,
-		func(_ INode, _ asymmetric.IPubKey, reqBytes []byte) ([]byte, error) {
+		func(_ context.Context, _ INode, _ asymmetric.IPubKey, reqBytes []byte) ([]byte, error) {
 			return []byte(fmt.Sprintf("echo: '%s'", string(reqBytes))), nil
 		},
 	)
 
+	ctx := context.Background()
+
 	msgBody := "hello, world!"
 	result, err := nodes[0].FetchPayload(
+		ctx,
 		nodes[1].GetMessageQueue().GetClient().GetPubKey(),
 		adapters.NewPayload(testutils.TcHead, []byte(msgBody)),
 	)
@@ -222,6 +231,7 @@ func TestFetchPayload(t *testing.T) {
 	}
 
 	_, err2 := nodes[0].FetchPayload(
+		ctx,
 		nodes[1].GetMessageQueue().GetClient().GetPubKey(),
 		adapters.NewPayload(testutils.TcHead, []byte(msgBody)),
 	)
@@ -245,15 +255,18 @@ func TestBroadcastPayload(t *testing.T) {
 	chResult := make(chan string)
 	nodes[1].HandleFunc(
 		testutils.TcHead,
-		func(_ INode, _ asymmetric.IPubKey, reqBytes []byte) ([]byte, error) {
+		func(_ context.Context, _ INode, _ asymmetric.IPubKey, reqBytes []byte) ([]byte, error) {
 			res := fmt.Sprintf("echo: '%s'", string(reqBytes))
 			go func() { chResult <- res }()
 			return nil, nil
 		},
 	)
 
+	ctx := context.Background()
+
 	msgBody := "hello, world!"
 	err := nodes[0].BroadcastPayload(
+		ctx,
 		nodes[1].GetMessageQueue().GetClient().GetPubKey(),
 		adapters.NewPayload(testutils.TcHead, []byte(msgBody)),
 	)
@@ -280,6 +293,7 @@ func TestBroadcastPayload(t *testing.T) {
 	}
 
 	err2 := nodes[0].BroadcastPayload(
+		ctx,
 		nodes[1].GetMessageQueue().GetClient().GetPubKey(),
 		adapters.NewPayload(testutils.TcHead, []byte(msgBody)),
 	)
@@ -377,21 +391,23 @@ func TestHandleWrapper(t *testing.T) {
 		return
 	}
 
+	ctx := context.Background()
+
 	sett := net_message.NewSettings(&net_message.SSettings{})
 	netMsg := node.testNewNetworkMessage(sett, msg)
-	if err := handler(nil, nil, netMsg); err != nil {
+	if err := handler(ctx, nil, nil, netMsg); err != nil {
 		t.Error(err)
 		return
 	}
 
-	if err := handler(nil, nil, netMsg); err != nil {
+	if err := handler(ctx, nil, nil, netMsg); err != nil {
 		t.Error("repeated message:", err.Error())
 		return
 	}
 
 	node.HandleFunc(
 		111,
-		func(_ INode, _ asymmetric.IPubKey, _ []byte) ([]byte, error) {
+		func(_ context.Context, _ INode, _ asymmetric.IPubKey, _ []byte) ([]byte, error) {
 			return nil, errors.New("some error")
 		},
 	)
@@ -409,7 +425,7 @@ func TestHandleWrapper(t *testing.T) {
 	}
 
 	netMsg2 := node.testNewNetworkMessage(sett, msg2)
-	if err := handler(nil, nil, netMsg2); err != nil {
+	if err := handler(ctx, nil, nil, netMsg2); err != nil {
 		t.Error(err) // works only logger
 		return
 	}
@@ -427,7 +443,7 @@ func TestHandleWrapper(t *testing.T) {
 	}
 
 	netMsg3 := node.testNewNetworkMessage(sett, msg3)
-	if err := handler(nil, nil, netMsg3); err != nil {
+	if err := handler(ctx, nil, nil, netMsg3); err != nil {
 		t.Error(err) // works only logger
 		return
 	}
@@ -445,7 +461,7 @@ func TestHandleWrapper(t *testing.T) {
 	}
 
 	netMsg4 := node.testNewNetworkMessage(sett, msg4)
-	if err := handler(nil, nil, netMsg4); err != nil {
+	if err := handler(ctx, nil, nil, netMsg4); err != nil {
 		t.Error(err) // works only logger
 		return
 	}
@@ -477,12 +493,13 @@ func TestStoreHashWithBroadcastMessage(t *testing.T) {
 	netMsg := node.testNewNetworkMessage(sett, msg)
 	logBuilder := anon_logger.NewLogBuilder("_")
 
-	if ok, err := node.storeHashWithBroadcast(logBuilder, netMsg); !ok || err != nil {
+	ctx := context.Background()
+	if ok, err := node.storeHashWithBroadcast(ctx, logBuilder, netMsg); !ok || err != nil {
 		t.Error(err)
 		return
 	}
 
-	if ok, err := node.storeHashWithBroadcast(logBuilder, netMsg); ok || err != nil {
+	if ok, err := node.storeHashWithBroadcast(ctx, logBuilder, netMsg); ok || err != nil {
 		switch {
 		case ok:
 			t.Error("success store one message again")
@@ -493,7 +510,7 @@ func TestStoreHashWithBroadcastMessage(t *testing.T) {
 	}
 
 	node.GetWrapperDB().Set(nil)
-	if ok, err := node.storeHashWithBroadcast(logBuilder, netMsg); ok || err == nil {
+	if ok, err := node.storeHashWithBroadcast(ctx, logBuilder, netMsg); ok || err == nil {
 		t.Error("success use store function with null database")
 		return
 	}
@@ -544,8 +561,10 @@ func TestRecvSendMessage(t *testing.T) {
 		return
 	}
 
+	ctx := context.Background()
+
 	for i := 0; i < testutils.TCQueueCapacity; i++ {
-		if err := node.send(msg); err != nil {
+		if err := node.send(ctx, msg); err != nil {
 			t.Error("failed send message (push to queue)")
 			return
 		}
@@ -553,7 +572,7 @@ func TestRecvSendMessage(t *testing.T) {
 
 	for i := 0; i < 3; i++ {
 		// message can be dequeued in the send's call time
-		if err := node.send(msg); err != nil {
+		if err := node.send(ctx, msg); err != nil {
 			return
 		}
 	}
@@ -584,40 +603,43 @@ func testNewNodes(t *testing.T, timeWait time.Duration, addresses [2]string, typ
 	for _, node := range nodes {
 		node.HandleFunc(
 			testutils.TcHead,
-			func(_ INode, _ asymmetric.IPubKey, reqBytes []byte) ([]byte, error) {
+			func(_ context.Context, _ INode, _ asymmetric.IPubKey, reqBytes []byte) ([]byte, error) {
 				// send response
 				return []byte(fmt.Sprintf("%s (response)", string(reqBytes))), nil
 			},
 		)
 	}
 
-	if err := nodes[2].GetNetworkNode().Listen(); err != nil {
-		t.Error(err)
-		return [5]INode{}, [5]context.CancelFunc{}
-	}
-	if err := nodes[4].GetNetworkNode().Listen(); err != nil {
-		t.Error(err)
-		return [5]INode{}, [5]context.CancelFunc{}
-	}
+	ctx := context.Background()
+	go func() {
+		if err := nodes[2].GetNetworkNode().Listen(ctx); err != nil {
+			t.Error(err)
+		}
+	}()
+	go func() {
+		if err := nodes[4].GetNetworkNode().Listen(ctx); err != nil {
+			t.Error(err)
+		}
+	}()
 
 	time.Sleep(200 * time.Millisecond)
 
 	// nodes to routes (nodes[0] -> nodes[2], nodes[1] -> nodes[4])
-	if err := nodes[0].GetNetworkNode().AddConnection(addresses[0]); err != nil {
+	if err := nodes[0].GetNetworkNode().AddConnection(ctx, addresses[0]); err != nil {
 		t.Error(err)
 		return [5]INode{}, [5]context.CancelFunc{}
 	}
-	if err := nodes[1].GetNetworkNode().AddConnection(addresses[1]); err != nil {
+	if err := nodes[1].GetNetworkNode().AddConnection(ctx, addresses[1]); err != nil {
 		t.Error(err)
 		return [5]INode{}, [5]context.CancelFunc{}
 	}
 
 	// routes to routes (nodes[3] -> nodes[2], nodes[3] -> nodes[4])
-	if err := nodes[3].GetNetworkNode().AddConnection(addresses[0]); err != nil {
+	if err := nodes[3].GetNetworkNode().AddConnection(ctx, addresses[0]); err != nil {
 		t.Error(err)
 		return [5]INode{}, [5]context.CancelFunc{}
 	}
-	if err := nodes[3].GetNetworkNode().AddConnection(addresses[1]); err != nil {
+	if err := nodes[3].GetNetworkNode().AddConnection(ctx, addresses[1]); err != nil {
 		t.Error(err)
 		return [5]INode{}, [5]context.CancelFunc{}
 	}

pkg/network/anonymity/examples/echo/main.go

@@ -59,7 +59,7 @@ func main() {
 		}
 	}()
 
-	service.HandleFunc(serviceHeader, func(_ anonymity.INode, _ asymmetric.IPubKey, reqBytes []byte) ([]byte, error) {
+	service.HandleFunc(serviceHeader, func(_ context.Context, _ anonymity.INode, _ asymmetric.IPubKey, reqBytes []byte) ([]byte, error) {
 		return []byte(fmt.Sprintf("echo: [%s]", string(reqBytes))), nil
 	})
 
@@ -70,7 +70,7 @@ func main() {
 	defer cancel1()
 
 	go func() { _ = service.Run(ctx1) }()
-	if err := service.GetNetworkNode().Listen(); err != nil {
+	if err := service.GetNetworkNode().Listen(ctx1); err != nil {
 		panic(err)
 	}
 	time.Sleep(time.Second) // wait
@@ -79,11 +79,12 @@ func main() {
 	defer cancel2()
 
 	go func() { _ = client.Run(ctx2) }()
-	if err := client.GetNetworkNode().AddConnection(serviceAddress); err != nil {
+	if err := client.GetNetworkNode().AddConnection(ctx2, serviceAddress); err != nil {
 		panic(err)
 	}
 
 	res, err := client.FetchPayload(
+		ctx2,
 		service.GetMessageQueue().GetClient().GetPubKey(),
 		adapters.NewPayload(serviceHeader, []byte("hello, world!")),
 	)

pkg/network/anonymity/examples/ping-ping/main.go

@@ -75,7 +75,7 @@ func main() {
 	defer cancel1()
 
 	go func() { _ = service1.Run(ctx1) }()
-	if err := service1.GetNetworkNode().Listen(); err != nil {
+	if err := service1.GetNetworkNode().Listen(ctx1); err != nil {
 		panic(err)
 	}
 	time.Sleep(time.Second)
@@ -84,11 +84,12 @@ func main() {
 	defer cancel2()
 
 	go func() { _ = service2.Run(ctx2) }()
-	if err := service2.GetNetworkNode().AddConnection(serviceAddress); err != nil {
+	if err := service2.GetNetworkNode().AddConnection(ctx2, serviceAddress); err != nil {
 		panic(err)
 	}
 
 	err := service2.BroadcastPayload(
+		ctx2,
 		service1.GetMessageQueue().GetClient().GetPubKey(),
 		adapters.NewPayload(
 			serviceHeader,
@@ -103,7 +104,7 @@ func main() {
 }
 
 func handler(serviceName string) anonymity.IHandlerF {
-	return func(node anonymity.INode, pubKey asymmetric.IPubKey, reqBytes []byte) ([]byte, error) {
+	return func(ctx context.Context, node anonymity.INode, pubKey asymmetric.IPubKey, reqBytes []byte) ([]byte, error) {
 		num, err := strconv.Atoi(string(reqBytes))
 		if err != nil {
 			panic(err)
@@ -117,6 +118,7 @@ func handler(serviceName string) anonymity.IHandlerF {
 		fmt.Printf("service '%s' got '%s#%d'\n", serviceName, val, num)
 
 		err = node.BroadcastPayload(
+			ctx,
 			pubKey,
 			adapters.NewPayload(
 				serviceHeader,

pkg/network/anonymity/queue/examples/main.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"time"
 
@@ -37,7 +38,7 @@ func main() {
 	defer cancel()
 
 	go func() {
-		if err := q.Run(ctx); err != nil {
+		if err := q.Run(ctx); err != nil && !errors.Is(err, context.Canceled) {
 			panic(err)
 		}
 	}()

pkg/network/anonymity/queue/queue.go

@@ -72,7 +72,7 @@ func (p *sMessageQueue) Run(pCtx context.Context) error {
 	for {
 		select {
 		case <-pCtx.Done():
-			return nil
+			return pCtx.Err()
 		case <-time.After(p.fSettings.GetDuration() / 2):
 			if p.poolHasLimit() {
 				continue
@@ -118,35 +118,19 @@ func (p *sMessageQueue) EnqueueMessage(pMsg message.IMessage) error {
 }
 
 func (p *sMessageQueue) DequeueMessage(pCtx context.Context) net_message.IMessage {
-	var (
-		result net_message.IMessage
-		closed = make(chan bool)
-	)
+	select {
+	case <-pCtx.Done():
+		return nil
+	case <-time.After(p.fSettings.GetDuration()):
+		p.fMutex.Lock()
+		queueLen := len(p.fQueue)
+		p.fMutex.Unlock()
 
-	go func() {
-		select {
-		case <-pCtx.Done():
-			closed <- true
-			return
-		case <-time.After(p.fSettings.GetDuration()):
-			defer func() { closed <- false }()
-
-			p.fMutex.Lock()
-			queueLen := len(p.fQueue)
-			p.fMutex.Unlock()
-
-			if queueLen == 0 {
-				result = <-p.fMsgPool.fQueue
-				return
-			}
-
-			result = <-p.fQueue
-			return
+		if queueLen == 0 {
+			return <-p.fMsgPool.fQueue
 		}
-	}()
-
-	<-closed
-	return result
+		return <-p.fQueue
+	}
 }
 
 func (p *sMessageQueue) newPseudoNetworkMessage() net_message.IMessage {

pkg/network/anonymity/queue/queue_test.go

@@ -3,6 +3,7 @@ package queue
 import (
 	"bytes"
 	"context"
+	"errors"
 	"fmt"
 	"testing"
 	"time"
@@ -72,7 +73,7 @@ func TestRunStopQueue(t *testing.T) {
 	defer cancel1()
 
 	go func() {
-		if err := queue.Run(ctx1); err != nil {
+		if err := queue.Run(ctx1); err != nil && !errors.Is(err, context.Canceled) {
 			t.Error(err)
 			return
 		}
@@ -156,7 +157,7 @@ func testQueue(queue IMessageQueue) error {
 	defer cancel()
 
 	go func() {
-		if err := queue.Run(ctx); err != nil {
+		if err := queue.Run(ctx); err != nil && !errors.Is(err, context.Canceled) {
 			return
 		}
 	}()

pkg/network/anonymity/types.go

@@ -1,6 +1,7 @@
 package anonymity
 
 import (
+	"context"
 	"time"
 
 	"github.com/number571/go-peer/pkg/crypto/asymmetric"
@@ -13,11 +14,12 @@ import (
 )
 
 type (
-	IHandlerF func(INode, asymmetric.IPubKey, []byte) ([]byte, error)
+	IHandlerF func(context.Context, INode, asymmetric.IPubKey, []byte) ([]byte, error)
 )
 
 type INode interface {
 	types.IRunner
+	HandleFunc(uint32, IHandlerF) INode
 
 	GetSettings() ISettings
 	GetWrapperDB() IWrapperDB
@@ -26,10 +28,8 @@ type INode interface {
 	GetListPubKeys() asymmetric.IListPubKeys
 	GetLogger() logger.ILogger
 
-	HandleFunc(uint32, IHandlerF) INode
-
-	BroadcastPayload(asymmetric.IPubKey, adapters.IPayload) error
-	FetchPayload(asymmetric.IPubKey, adapters.IPayload) ([]byte, error)
+	BroadcastPayload(context.Context, asymmetric.IPubKey, adapters.IPayload) error
+	FetchPayload(context.Context, asymmetric.IPubKey, adapters.IPayload) ([]byte, error)
 }
 
 type ISettings interface {

pkg/network/conn/conn.go

@@ -2,6 +2,7 @@ package conn
 
 import (
 	"bytes"
+	"context"
 	"errors"
 	"fmt"
 	"net"
@@ -83,7 +84,7 @@ func (p *sConn) Close() error {
 	return p.fSocket.Close()
 }
 
-func (p *sConn) WriteMessage(pMsg message.IMessage) error {
+func (p *sConn) WriteMessage(pCtx context.Context, pMsg message.IMessage) error {
 	p.fMutex.Lock()
 	defer p.fMutex.Unlock()
 
@@ -93,7 +94,7 @@ func (p *sConn) WriteMessage(pMsg message.IMessage) error {
 	voidBytes := prng.GetBytes(randVoidSize)
 	encMsgBytes := p.getCipher().EncryptBytes(pMsg.ToBytes())
 
-	err := p.sendBytes(bytes.Join(
+	err := p.sendBytes(pCtx, bytes.Join(
 		[][]byte{
 			p.getHeadBytes(encMsgBytes, voidBytes),
 			bytes.Join(
@@ -113,14 +114,14 @@ func (p *sConn) WriteMessage(pMsg message.IMessage) error {
 	return nil
 }
 
-func (p *sConn) ReadMessage(pChRead chan struct{}) (message.IMessage, error) {
+func (p *sConn) ReadMessage(pCtx context.Context, pChRead chan struct{}) (message.IMessage, error) {
 	// large wait read deadline => the connection has not sent anything yet
-	encMsgSize, voidSize, gotHash, err := p.recvHeadBytes(pChRead, p.fSettings.GetWaitReadDeadline())
+	encMsgSize, voidSize, gotHash, err := p.recvHeadBytes(pCtx, pChRead, p.fSettings.GetWaitReadDeadline())
 	if err != nil {
 		return nil, fmt.Errorf("receive head bytes: %w", err)
 	}
 
-	dataBytes, err := p.recvDataBytes(encMsgSize + voidSize)
+	dataBytes, err := p.recvDataBytes(pCtx, encMsgSize+voidSize)
 	if err != nil {
 		return nil, fmt.Errorf("receive data bytes: %w", err)
 	}
@@ -147,18 +148,23 @@ func (p *sConn) ReadMessage(pChRead chan struct{}) (message.IMessage, error) {
 	return msg, nil
 }
 
-func (p *sConn) sendBytes(pBytes []byte) error {
+func (p *sConn) sendBytes(pCtx context.Context, pBytes []byte) error {
 	bytesPtr := uint64(len(pBytes))
 	for bytesPtr != 0 {
-		p.fSocket.SetWriteDeadline(time.Now().Add(p.fSettings.GetWriteDeadline()))
+		select {
+		case <-pCtx.Done():
+			return pCtx.Err()
+		default:
+			p.fSocket.SetWriteDeadline(time.Now().Add(p.fSettings.GetWriteDeadline()))
 
-		n, err := p.fSocket.Write(pBytes[:bytesPtr])
-		if err != nil {
-			return fmt.Errorf("write tcp bytes: %w", err)
+			n, err := p.fSocket.Write(pBytes[:bytesPtr])
+			if err != nil {
+				return fmt.Errorf("write tcp bytes: %w", err)
+			}
+
+			bytesPtr = bytesPtr - uint64(n)
+			pBytes = pBytes[:bytesPtr]
 		}
-
-		bytesPtr = bytesPtr - uint64(n)
-		pBytes = pBytes[:bytesPtr]
 	}
 	return nil
 }
@@ -190,10 +196,8 @@ func (p *sConn) getHeadBytes(pEncMsgBytes, pVoidBytes []byte) []byte {
 	))
 }
 
-func (p *sConn) recvHeadBytes(pChRead chan struct{}, deadline time.Duration) (uint64, uint64, []byte, error) {
-	defer func() {
-		pChRead <- struct{}{}
-	}()
+func (p *sConn) recvHeadBytes(pCtx context.Context, pChRead chan<- struct{}, deadline time.Duration) (uint64, uint64, []byte, error) {
+	defer func() { pChRead <- struct{}{} }()
 
 	const (
 		firstSizeIndex  = encoding.CSizeUint64
@@ -203,15 +207,28 @@ func (p *sConn) recvHeadBytes(pChRead chan struct{}, deadline time.Duration) (ui
 	)
 
 	p.fSocket.SetReadDeadline(time.Now().Add(deadline))
-
 	encRecvHead := make([]byte, cEncryptRecvHeadSize)
-	n, err := p.fSocket.Read(encRecvHead)
-	if err != nil {
-		return 0, 0, nil, fmt.Errorf("read tcp header block: %w", err)
-	}
 
-	if n != cEncryptRecvHeadSize {
-		return 0, 0, nil, errors.New("invalid header block")
+	chErr := make(chan error)
+	go func() {
+		n, err := p.fSocket.Read(encRecvHead)
+		if err != nil {
+			chErr <- fmt.Errorf("read tcp header block: %w", err)
+		}
+		if n != cEncryptRecvHeadSize {
+			chErr <- errors.New("invalid header block")
+		}
+		chErr <- nil
+	}()
+
+	select {
+	case <-pCtx.Done():
+		return 0, 0, nil, pCtx.Err()
+	case err := <-chErr:
+		if err != nil {
+			return 0, 0, nil, err
+		}
+		break
 	}
 
 	recvHead := p.getCipher().DecryptBytes(encRecvHead)
@@ -251,28 +268,33 @@ func (p *sConn) recvHeadBytes(pChRead chan struct{}, deadline time.Duration) (ui
 	return encMsgSize, voidSize, recvHead[firstHashIndex:secondHashIndex], nil
 }
 
-func (p *sConn) recvDataBytes(pMustLen uint64) ([]byte, error) {
+func (p *sConn) recvDataBytes(pCtx context.Context, pMustLen uint64) ([]byte, error) {
 	dataRaw := make([]byte, 0, pMustLen)
 
 	mustLen := pMustLen
 	for mustLen != 0 {
-		p.fSocket.SetReadDeadline(time.Now().Add(p.fSettings.GetReadDeadline()))
+		select {
+		case <-pCtx.Done():
+			return nil, pCtx.Err()
+		default:
+			p.fSocket.SetReadDeadline(time.Now().Add(p.fSettings.GetReadDeadline()))
 
-		buffer := make([]byte, mustLen)
-		n, err := p.fSocket.Read(buffer)
-		if err != nil {
-			return nil, err
+			buffer := make([]byte, mustLen)
+			n, err := p.fSocket.Read(buffer)
+			if err != nil {
+				return nil, err
+			}
+
+			dataRaw = bytes.Join(
+				[][]byte{
+					dataRaw,
+					buffer[:n],
+				},
+				[]byte{},
+			)
+
+			mustLen -= uint64(n)
 		}
-
-		dataRaw = bytes.Join(
-			[][]byte{
-				dataRaw,
-				buffer[:n],
-			},
-			[]byte{},
-		)
-
-		mustLen -= uint64(n)
 	}
 
 	return dataRaw, nil

pkg/network/conn/conn_test.go

@@ -2,6 +2,7 @@ package conn
 
 import (
 	"bytes"
+	"context"
 	"net"
 	"strings"
 	"testing"
@@ -108,7 +109,9 @@ func TestClosedConn(t *testing.T) {
 
 	pld := payload.NewPayload(1, []byte("aaa"))
 	msg := message.NewMessage(conn.GetSettings(), pld)
-	if err := conn.WriteMessage(msg); err == nil {
+
+	ctx := context.Background()
+	if err := conn.WriteMessage(ctx, msg); err == nil {
 		t.Error("success write payload to closed connection")
 		return
 	}
@@ -116,18 +119,18 @@ func TestClosedConn(t *testing.T) {
 	readCh := make(chan struct{})
 	go func() { <-readCh }()
 
-	if _, err := conn.ReadMessage(readCh); err == nil {
+	if _, err := conn.ReadMessage(ctx, readCh); err == nil {
 		t.Error("success read payload from closed connection")
 		return
 	}
 
 	sconn := conn.(*sConn)
-	if err := sconn.sendBytes([]byte("hello, world!")); err == nil {
+	if err := sconn.sendBytes(ctx, []byte("hello, world!")); err == nil {
 		t.Error("success send bytes to closed connection")
 		return
 	}
 
-	if _, err := sconn.recvDataBytes(128); err == nil {
+	if _, err := sconn.recvDataBytes(ctx, 128); err == nil {
 		t.Error("success recv data bytes from closed connection")
 		return
 	}
@@ -135,7 +138,7 @@ func TestClosedConn(t *testing.T) {
 	readCh2 := make(chan struct{})
 	go func() { <-readCh2 }()
 
-	if _, _, _, err := sconn.recvHeadBytes(readCh2, time.Minute); err == nil {
+	if _, _, _, err := sconn.recvHeadBytes(ctx, readCh2, time.Minute); err == nil {
 		t.Error("success recv head bytes from closed connection")
 		return
 	}
@@ -197,7 +200,8 @@ func testConn(t *testing.T, pAddr, pNetworkKey string) {
 
 	pld := payload.NewPayload(tcHead, []byte(tcBody))
 	msg := message.NewMessage(conn.GetSettings(), pld)
-	if err := conn.WriteMessage(msg); err != nil {
+	ctx := context.Background()
+	if err := conn.WriteMessage(ctx, msg); err != nil {
 		t.Error(err)
 		return
 	}
@@ -205,7 +209,7 @@ func testConn(t *testing.T, pAddr, pNetworkKey string) {
 	readCh := make(chan struct{})
 	go func() { <-readCh }()
 
-	msgRecv, err := conn.ReadMessage(readCh)
+	msgRecv, err := conn.ReadMessage(ctx, readCh)
 	if err != nil {
 		t.Error(err)
 		return
@@ -247,14 +251,16 @@ func testNewService(t *testing.T, pAddr, pNetworkKey string) net.Listener {
 			readCh := make(chan struct{})
 			go func() { <-readCh }()
 
-			msg, err := conn.ReadMessage(readCh)
+			ctx := context.Background()
+
+			msg, err := conn.ReadMessage(ctx, readCh)
 			if err != nil {
 				break
 			}
 
 			ok := func() bool {
 				defer conn.Close()
-				return conn.WriteMessage(msg) == nil
+				return conn.WriteMessage(ctx, msg) == nil
 			}()
 
 			if !ok {

pkg/network/conn/types.go

@@ -1,6 +1,7 @@
 package conn
 
 import (
+	"context"
 	"net"
 	"time"
 
@@ -14,8 +15,8 @@ type IConn interface {
 	GetSettings() ISettings
 	GetSocket() net.Conn
 
-	WriteMessage(net_message.IMessage) error
-	ReadMessage(chan struct{}) (net_message.IMessage, error)
+	WriteMessage(context.Context, net_message.IMessage) error
+	ReadMessage(context.Context, chan struct{}) (net_message.IMessage, error)
 }
 
 type ISettings interface {

pkg/network/conn_keeper/conn_keeper.go

@@ -47,17 +47,17 @@ func (p *sConnKeeper) Run(pCtx context.Context) error {
 	}()
 
 	for {
-		p.tryConnectToAll()
+		p.tryConnectToAll(pCtx)
 		select {
 		case <-pCtx.Done():
-			return nil
+			return pCtx.Err()
 		case <-time.After(p.fSettings.GetDuration()):
 			// next iter
 		}
 	}
 }
 
-func (p *sConnKeeper) tryConnectToAll() {
+func (p *sConnKeeper) tryConnectToAll(pCtx context.Context) {
 	connList := p.fSettings.GetConnections()
 
 	wg := sync.WaitGroup{}
@@ -72,7 +72,7 @@ func (p *sConnKeeper) tryConnectToAll() {
 				return
 			}
 
-			p.fNode.AddConnection(addr)
+			p.fNode.AddConnection(pCtx, addr)
 		}(addr)
 	}
 

pkg/network/conn_keeper/conn_keeper_test.go

@@ -2,6 +2,7 @@ package conn_keeper
 
 import (
 	"context"
+	"errors"
 	"net"
 	"testing"
 	"time"
@@ -68,7 +69,7 @@ func TestConnKeeper(t *testing.T) {
 	defer cancel1()
 
 	go func() {
-		if err := connKeeper.Run(ctx1); err != nil {
+		if err := connKeeper.Run(ctx1); err != nil && !errors.Is(err, context.Canceled) {
 			t.Error(err)
 			return
 		}

pkg/network/examples/broadcast/main.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"context"
 	"fmt"
 	"time"
 
@@ -34,12 +35,13 @@ func main() {
 	service1.HandleFunc(serviceHeader, handler("#1"))
 	service2.HandleFunc(serviceHeader, handler("#2"))
 
-	if err := service1.Listen(); err != nil {
+	ctx := context.Background()
+	if err := service1.Listen(ctx); err != nil {
 		panic(err)
 	}
 	time.Sleep(time.Second) // wait
 
-	if err := service2.AddConnection(serviceAddress); err != nil {
+	if err := service2.AddConnection(ctx, serviceAddress); err != nil {
 		panic(err)
 	}
 
@@ -53,7 +55,7 @@ func main() {
 		}
 	}()
 
-	err = conn.WriteMessage(message.NewMessage(
+	err = conn.WriteMessage(ctx, message.NewMessage(
 		conn.GetSettings(),
 		payload.NewPayload(
 			serviceHeader,
@@ -68,8 +70,8 @@ func main() {
 }
 
 func handler(serviceName string) network.IHandlerF {
-	return func(n network.INode, _ conn.IConn, msg message.IMessage) error {
-		defer n.BroadcastMessage(msg)
+	return func(ctx context.Context, n network.INode, _ conn.IConn, msg message.IMessage) error {
+		defer n.BroadcastMessage(ctx, msg)
 		fmt.Printf("service '%s' got '%s'\n", serviceName, string(msg.GetPayload().GetBody()))
 		return nil
 	}

pkg/network/examples/echo/main.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"context"
 	"fmt"
 	"time"
 
@@ -29,7 +30,8 @@ func main() {
 
 	service.HandleFunc(serviceHeader, handler())
 
-	if err := service.Listen(); err != nil {
+	ctx := context.Background()
+	if err := service.Listen(ctx); err != nil {
 		panic(err)
 	}
 	time.Sleep(time.Second) // wait
@@ -48,14 +50,14 @@ func main() {
 		conn.GetSettings(),
 		payload.NewPayload(serviceHeader, []byte("hello, world!")),
 	)
-	if err := conn.WriteMessage(sendMsg); err != nil {
+	if err := conn.WriteMessage(ctx, sendMsg); err != nil {
 		panic(err)
 	}
 
 	readCh := make(chan struct{})
 	go func() { <-readCh }()
 
-	recvMsg, err := conn.ReadMessage(readCh)
+	recvMsg, err := conn.ReadMessage(ctx, readCh)
 	if err != nil {
 		panic(err)
 	}
@@ -64,8 +66,8 @@ func main() {
 }
 
 func handler() network.IHandlerF {
-	return func(node network.INode, c conn.IConn, msg message.IMessage) error {
-		c.WriteMessage(message.NewMessage(
+	return func(ctx context.Context, node network.INode, c conn.IConn, msg message.IMessage) error {
+		c.WriteMessage(ctx, message.NewMessage(
 			node.GetSettings().GetConnSettings(),
 			payload.NewPayload(
 				serviceHeader,

pkg/network/examples/ping-pong/main.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"context"
 	"fmt"
 	"strconv"
 	"time"
@@ -34,13 +35,14 @@ func main() {
 	service1.HandleFunc(serviceHeader, handler("#1"))
 	service2.HandleFunc(serviceHeader, handler("#2"))
 
-	if err := service1.Listen(); err != nil {
+	ctx := context.Background()
+	if err := service1.Listen(ctx); err != nil {
 		panic(err)
 	}
 
 	time.Sleep(time.Second) // wait
 
-	if err := service2.AddConnection(serviceAddress); err != nil {
+	if err := service2.AddConnection(ctx, serviceAddress); err != nil {
 		panic(err)
 	}
 
@@ -51,13 +53,13 @@ func main() {
 			[]byte("0"),
 		),
 	)
-	service2.BroadcastMessage(msg)
+	service2.BroadcastMessage(ctx, msg)
 
 	select {}
 }
 
 func handler(serviceName string) network.IHandlerF {
-	return func(n network.INode, _ conn.IConn, msg message.IMessage) error {
+	return func(ctx context.Context, n network.INode, _ conn.IConn, msg message.IMessage) error {
 		time.Sleep(time.Second) // delay for view "ping-pong" game
 
 		num, err := strconv.Atoi(string(msg.GetPayload().GetBody()))
@@ -71,7 +73,7 @@ func handler(serviceName string) network.IHandlerF {
 		}
 
 		fmt.Printf("service '%s' got '%s#%d'\n", serviceName, val, num)
-		n.BroadcastMessage(message.NewMessage(
+		n.BroadcastMessage(ctx, message.NewMessage(
 			n.GetSettings().GetConnSettings(),
 			payload.NewPayload(
 				serviceHeader,

pkg/network/network.go

@@ -1,6 +1,7 @@
 package network
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"net"
@@ -48,7 +49,7 @@ func (p *sNode) GetSettings() ISettings {
 }
 
 // Puts the hash of the message in the buffer and sends the message to all connections of the node.
-func (p *sNode) BroadcastMessage(pMsg message.IMessage) error {
+func (p *sNode) BroadcastMessage(pCtx context.Context, pMsg message.IMessage) error {
 	// node can redirect received message
 	_ = p.fQueuePusher.Push(pMsg.GetHash(), []byte{})
 
@@ -61,7 +62,7 @@ func (p *sNode) BroadcastMessage(pMsg message.IMessage) error {
 
 		chErr := make(chan error)
 		go func(c conn.IConn) {
-			chErr <- c.WriteMessage(pMsg)
+			chErr <- c.WriteMessage(pCtx, pMsg)
 		}(c)
 
 		go func(a string, c conn.IConn) {
@@ -95,17 +96,19 @@ func (p *sNode) BroadcastMessage(pMsg message.IMessage) error {
 // Opens a tcp connection to receive data from outside.
 // Checks the number of valid connections.
 // Redirects connections to the handle router.
-func (p *sNode) Listen() error {
+func (p *sNode) Listen(pCtx context.Context) error {
 	listener, err := net.Listen("tcp", p.fSettings.GetAddress())
 	if err != nil {
 		return fmt.Errorf("run node: %w", err)
 	}
+	defer listener.Close()
 
-	go func(pListener net.Listener) {
-		defer pListener.Close()
-		p.setListener(pListener)
-
-		for {
+	p.setListener(listener)
+	for {
+		select {
+		case <-pCtx.Done():
+			return pCtx.Err()
+		default:
 			tconn, err := p.getListener().Accept()
 			if err != nil {
 				break
@@ -121,11 +124,9 @@ func (p *sNode) Listen() error {
 			address := tconn.RemoteAddr().String()
 
 			p.setConnection(address, conn)
-			go p.handleConn(address, conn)
+			go p.handleConn(pCtx, address, conn)
 		}
-	}(listener)
-
-	return nil
+	}
 }
 
 // Closes the listener and all connections.
@@ -170,7 +171,7 @@ func (p *sNode) GetConnections() map[string]conn.IConn {
 
 // Connects to the node at the specified address and automatically starts reading all incoming messages.
 // Checks the number of connections.
-func (p *sNode) AddConnection(pAddress string) error {
+func (p *sNode) AddConnection(pCtx context.Context, pAddress string) error {
 	if p.hasMaxConnSize() {
 		return errors.New("has max connections size")
 	}
@@ -186,7 +187,7 @@ func (p *sNode) AddConnection(pAddress string) error {
 	}
 
 	p.setConnection(pAddress, conn)
-	go p.handleConn(pAddress, conn)
+	go p.handleConn(pCtx, pAddress, conn)
 
 	return nil
 }
@@ -211,7 +212,7 @@ func (p *sNode) DelConnection(pAddress string) error {
 }
 
 // Processes the received data from the connection.
-func (p *sNode) handleConn(pAddress string, pConn conn.IConn) {
+func (p *sNode) handleConn(pCtx context.Context, pAddress string, pConn conn.IConn) {
 	defer p.DelConnection(pAddress)
 	for {
 		var (
@@ -220,23 +221,30 @@ func (p *sNode) handleConn(pAddress string, pConn conn.IConn) {
 		)
 
 		go func() {
-			msg, err := pConn.ReadMessage(readerCh)
+			msg, err := pConn.ReadMessage(pCtx, readerCh)
 			if err != nil {
 				returnCh <- false
 				return
 			}
-			returnCh <- p.handleMessage(pConn, msg)
+			returnCh <- p.handleMessage(pCtx, pConn, msg)
 		}()
 
-		<-readerCh
 		select {
+		case <-pCtx.Done():
+			return
+		case <-readerCh:
+			// pass
+		}
+
+		select {
+		case <-pCtx.Done():
+			return
+		case <-time.After(p.fSettings.GetReadTimeout()):
+			return
 		case ok := <-returnCh:
 			if !ok {
 				return
 			}
-		case <-time.After(p.fSettings.GetReadTimeout()):
-			<-returnCh
-			return
 		}
 	}
 }
@@ -244,7 +252,7 @@ func (p *sNode) handleConn(pAddress string, pConn conn.IConn) {
 // Processes the message for correctness and redirects it to the handler function.
 // Returns true if the message was successfully redirected to the handler function
 // > or if the message already existed in the hash value store.
-func (p *sNode) handleMessage(pConn conn.IConn, pMsg message.IMessage) bool {
+func (p *sNode) handleMessage(pCtx context.Context, pConn conn.IConn, pMsg message.IMessage) bool {
 	// hash of message already in queue
 	if !p.fQueuePusher.Push(pMsg.GetHash(), []byte{}) {
 		return true
@@ -258,7 +266,7 @@ func (p *sNode) handleMessage(pConn conn.IConn, pMsg message.IMessage) bool {
 		return false
 	}
 
-	if err := f(p, pConn, pMsg); err != nil {
+	if err := f(pCtx, p, pConn, pMsg); err != nil {
 		// function error = protocol error
 		return false
 	}

pkg/network/network_test.go

@@ -1,6 +1,7 @@
 package network
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"sync"
@@ -97,9 +98,9 @@ func TestBroadcast(t *testing.T) {
 	wg.Add(4 * tcIter)
 
 	headHandle := uint64(testutils.TcHead)
-	handleF := func(node INode, conn conn.IConn, pMsg message.IMessage) error {
+	handleF := func(pCtx context.Context, node INode, conn conn.IConn, pMsg message.IMessage) error {
 		defer wg.Done()
-		defer node.BroadcastMessage(pMsg)
+		defer node.BroadcastMessage(pCtx, pMsg)
 
 		tcMutex.Lock()
 		defer tcMutex.Unlock()
@@ -127,6 +128,7 @@ func TestBroadcast(t *testing.T) {
 	}
 
 	// nodes[0] -> nodes[1:]
+	ctx := context.Background()
 	for i := 0; i < tcIter; i++ {
 		go func(i int) {
 			pld := payload.NewPayload(
@@ -134,7 +136,7 @@ func TestBroadcast(t *testing.T) {
 				[]byte(fmt.Sprintf(testutils.TcBodyTemplate, i)),
 			)
 			sett := nodes[0].GetSettings().GetConnSettings()
-			nodes[0].BroadcastMessage(message.NewMessage(sett, pld))
+			nodes[0].BroadcastMessage(ctx, message.NewMessage(sett, pld))
 		}(i)
 	}
 
@@ -228,49 +230,60 @@ func TestNodeConnection(t *testing.T) {
 	)
 	defer testFreeNodes([]INode{node1, node2, node3})
 
-	if err := node2.Listen(); err != nil {
-		t.Error(err)
-		return
-	}
+	ctx := context.Background()
+	go func() {
+		if err := node2.Listen(ctx); err != nil {
+			t.Error(err)
+			return
+		}
+	}()
 	defer node2.Close()
 
-	if err := node2.Listen(); err == nil {
-		t.Error("success second run node")
-		return
-	}
+	time.Sleep(200 * time.Millisecond)
 
-	if err := node3.Listen(); err != nil {
-		t.Error(err)
-		return
-	}
+	go func() {
+		if err := node2.Listen(ctx); err == nil {
+			t.Error("success second run node")
+			return
+		}
+	}()
+
+	go func() {
+		if err := node3.Listen(ctx); err != nil {
+			t.Error(err)
+			return
+		}
+	}()
 	defer node3.Close()
 
-	if err := node1.AddConnection("unknown_connection_address"); err == nil {
+	time.Sleep(200 * time.Millisecond)
+
+	if err := node1.AddConnection(ctx, "unknown_connection_address"); err == nil {
 		t.Error("success add incorrect connection address")
 		return
 	}
 
-	if err := node1.AddConnection(testutils.TgAddrs[27]); err != nil {
+	if err := node1.AddConnection(ctx, testutils.TgAddrs[27]); err != nil {
 		t.Error(err)
 		return
 	}
 
-	if err := node1.AddConnection(testutils.TgAddrs[27]); err == nil {
+	if err := node1.AddConnection(ctx, testutils.TgAddrs[27]); err == nil {
 		t.Error("success add already exist connection")
 		return
 	}
 
-	if err := node1.AddConnection(testutils.TgAddrs[28]); err != nil {
+	if err := node1.AddConnection(ctx, testutils.TgAddrs[28]); err != nil {
 		t.Error(err)
 		return
 	}
 
-	if err := node1.AddConnection(testutils.TgAddrs[28]); err == nil {
+	if err := node1.AddConnection(ctx, testutils.TgAddrs[28]); err == nil {
 		t.Error("success add second connection with limit = 1")
 		return
 	}
 
-	if err := node3.AddConnection(testutils.TgAddrs[27]); err != nil {
+	if err := node3.AddConnection(ctx, testutils.TgAddrs[27]); err != nil {
 		t.Error(err)
 		return
 	}
@@ -295,11 +308,6 @@ func TestNodeConnection(t *testing.T) {
 		t.Error(err)
 		return
 	}
-
-	if err := node2.Close(); err == nil {
-		t.Error("success stop already stopped process")
-		return
-	}
 }
 
 func TestHandleMessage(t *testing.T) {
@@ -309,28 +317,29 @@ func TestHandleMessage(t *testing.T) {
 	defer testFreeNodes([]INode{node})
 
 	sett := node.GetSettings().GetConnSettings()
+	ctx := context.Background()
 
 	node.HandleFunc(1, nil)
 	msg1 := message.NewMessage(sett, payload.NewPayload(1, []byte{1}))
-	if ok := node.handleMessage(nil, msg1); ok {
+	if ok := node.handleMessage(ctx, nil, msg1); ok {
 		t.Error("success handle message with nil function")
 		return
 	}
 
-	node.HandleFunc(1, func(i1 INode, i2 conn.IConn, b message.IMessage) error {
+	node.HandleFunc(1, func(ctx context.Context, i1 INode, i2 conn.IConn, b message.IMessage) error {
 		return errors.New("some error")
 	})
 	msg2 := message.NewMessage(sett, payload.NewPayload(1, []byte{2}))
-	if ok := node.handleMessage(nil, msg2); ok {
+	if ok := node.handleMessage(ctx, nil, msg2); ok {
 		t.Error("success handle message with got error from function")
 		return
 	}
 
-	node.HandleFunc(1, func(i1 INode, i2 conn.IConn, b message.IMessage) error {
+	node.HandleFunc(1, func(ctx context.Context, i1 INode, i2 conn.IConn, b message.IMessage) error {
 		return nil
 	})
 	msg3 := message.NewMessage(sett, payload.NewPayload(1, []byte{3}))
-	if ok := node.handleMessage(nil, msg3); !ok {
+	if ok := node.handleMessage(ctx, nil, msg3); !ok {
 		t.Error("failed handle message with correct function")
 		return
 	}
@@ -353,20 +362,18 @@ func testNodes() ([5]INode, map[INode]map[string]bool, error) {
 		nodes[i] = newTestNode(addrs[i], testutils.TCMaxConnects, time.Minute)
 	}
 
-	if err := nodes[2].Listen(); err != nil {
-		return nodes, nil, err
-	}
-	if err := nodes[4].Listen(); err != nil {
-		return nodes, nil, err
-	}
+	ctx := context.Background()
+
+	go func() { _ = nodes[2].Listen(ctx) }()
+	go func() { _ = nodes[4].Listen(ctx) }()
 
 	time.Sleep(500 * time.Millisecond)
 
-	nodes[0].AddConnection(testutils.TgAddrs[0])
-	nodes[1].AddConnection(testutils.TgAddrs[1])
+	nodes[0].AddConnection(ctx, testutils.TgAddrs[0])
+	nodes[1].AddConnection(ctx, testutils.TgAddrs[1])
 
-	nodes[3].AddConnection(testutils.TgAddrs[0])
-	nodes[3].AddConnection(testutils.TgAddrs[1])
+	nodes[3].AddConnection(ctx, testutils.TgAddrs[0])
+	nodes[3].AddConnection(ctx, testutils.TgAddrs[1])
 
 	mapp := make(map[INode]map[string]bool)
 	for _, node := range nodes {

pkg/network/types.go

@@ -1,6 +1,7 @@
 package network
 
 import (
+	"context"
 	"time"
 
 	"github.com/number571/go-peer/pkg/network/conn"
@@ -9,21 +10,21 @@ import (
 )
 
 type (
-	IHandlerF func(INode, conn.IConn, message.IMessage) error
+	IHandlerF func(context.Context, INode, conn.IConn, message.IMessage) error
 )
 
 type INode interface {
 	types.ICloser
-	Listen() error
+	Listen(context.Context) error
 
 	GetSettings() ISettings
 	GetConnections() map[string]conn.IConn
 
-	AddConnection(string) error
+	AddConnection(context.Context, string) error
 	DelConnection(string) error
 
 	HandleFunc(uint64, IHandlerF) INode
-	BroadcastMessage(message.IMessage) error
+	BroadcastMessage(context.Context, message.IMessage) error
 }
 
 type ISettings interface {

test/result/badge.svg

@@ -1 +1 @@
-<svg xmlns="http://www.w3.org/2000/svg" xmlns:xlink="http://www.w3.org/1999/xlink" width="96" height="20" role="img" aria-label="coverage: 45%"><title>coverage: 45%</title><linearGradient id="s" x2="0" y2="100%"><stop offset="0" stop-color="#bbb" stop-opacity=".1"/><stop offset="1" stop-opacity=".1"/></linearGradient><clipPath id="r"><rect width="96" height="20" rx="3" fill="#fff"/></clipPath><g clip-path="url(#r)"><rect width="61" height="20" fill="#555"/><rect x="61" width="35" height="20" fill="#e05d44"/><rect width="96" height="20" fill="url(#s)"/></g><g fill="#fff" text-anchor="middle" font-family="Verdana,Geneva,DejaVu Sans,sans-serif" text-rendering="geometricPrecision" font-size="110"><text aria-hidden="true" x="315" y="150" fill="#010101" fill-opacity=".3" transform="scale(.1)" textLength="510">coverage</text><text x="315" y="140" transform="scale(.1)" fill="#fff" textLength="510">coverage</text><text aria-hidden="true" x="775" y="150" fill="#010101" fill-opacity=".3" transform="scale(.1)" textLength="250">45%</text><text x="775" y="140" transform="scale(.1)" fill="#fff" textLength="250">45%</text></g></svg>
+<svg xmlns="http://www.w3.org/2000/svg" xmlns:xlink="http://www.w3.org/1999/xlink" width="96" height="20" role="img" aria-label="coverage: 46%"><title>coverage: 46%</title><linearGradient id="s" x2="0" y2="100%"><stop offset="0" stop-color="#bbb" stop-opacity=".1"/><stop offset="1" stop-opacity=".1"/></linearGradient><clipPath id="r"><rect width="96" height="20" rx="3" fill="#fff"/></clipPath><g clip-path="url(#r)"><rect width="61" height="20" fill="#555"/><rect x="61" width="35" height="20" fill="#e05d44"/><rect width="96" height="20" fill="url(#s)"/></g><g fill="#fff" text-anchor="middle" font-family="Verdana,Geneva,DejaVu Sans,sans-serif" text-rendering="geometricPrecision" font-size="110"><text aria-hidden="true" x="315" y="150" fill="#010101" fill-opacity=".3" transform="scale(.1)" textLength="510">coverage</text><text x="315" y="140" transform="scale(.1)" fill="#fff" textLength="510">coverage</text><text aria-hidden="true" x="775" y="150" fill="#010101" fill-opacity=".3" transform="scale(.1)" textLength="250">46%</text><text x="775" y="140" transform="scale(.1)" fill="#fff" textLength="250">46%</text></g></svg>