fixes

Author: asmyasnikov

coordination/coordination.go renamed to coordination/client.go

@@ -3,7 +3,6 @@ package coordination
 import (
 	"context"
 	"encoding/binary"
-	"fmt"
 	"math"
 	"math/rand"
 	"sync"
@@ -35,7 +34,7 @@ type (
 		sessionReconnectDelay   time.Duration
 		trace                   *trace.Coordination
 
-		ctx               context.Context
+		ctx               context.Context //nolint:containedctx
 		cancel            context.CancelFunc
 		sessionClosedChan chan struct{}
 		controller        *conversation.Controller
@@ -54,7 +53,7 @@ type (
 type lease struct {
 	session *session
 	name    string
-	ctx     context.Context
+	ctx     context.Context //nolint:containedctx
 	cancel  context.CancelFunc
 }
 
@@ -131,84 +130,8 @@ func (s *session) updateCancelStream(cancel context.CancelFunc) {
 	s.cancelStream = cancel
 }
 
-type sessionStream struct {
-	sessionID    uint64
-	stream       Ydb_Coordination_V1.CoordinationService_SessionClient
-	cancelStream context.CancelFunc
-}
-
-func newSessionStream(
-	ctx context.Context,
-	client Ydb_Coordination_V1.CoordinationServiceClient,
-	t *trace.Coordination,
-) (
-	_ *sessionStream, finalErr error,
-) {
-	streamCtx, streamCancel := xcontext.WithCancel(xcontext.ValueOnly(ctx))
-	defer func() {
-		if finalErr != nil {
-			streamCancel()
-		}
-	}()
-
-	onDone := trace.CoordinationOnNewSessionClient(t, &ctx,
-		stack.FunctionID("github.com/ydb-platform/ydb-go-sdk/3/internal/coordination.newSessionStream"),
-	)
-	defer func() {
-		onDone(finalErr)
-	}()
-
-	stream, err := client.Session(streamCtx)
-	if err != nil {
-		return nil, xerrors.WithStackTrace(err)
-	}
-
-	err = stream.Send(&Ydb_Coordination.SessionRequest{
-		Request: &Ydb_Coordination.SessionRequest_SessionStart_{
-			SessionStart: &Ydb_Coordination.SessionRequest_SessionStart{
-				Path:          "/a/b/c",
-				ProtectionKey: protectionKey,
-			},
-		},
-	})
-	if err != nil {
-		return nil, xerrors.WithStackTrace(err)
-	}
-
-	for {
-		msg, err := stream.Recv()
-		if err != nil {
-			return nil, xerrors.WithStackTrace(err)
-		}
-
-		switch t := msg.GetResponse().(type) {
-		case *Ydb_Coordination.SessionResponse_SessionStarted_:
-			return &sessionStream{
-				sessionID:    t.SessionStarted.GetSessionId(),
-				stream:       stream,
-				cancelStream: streamCancel,
-			}, nil
-		case *Ydb_Coordination.SessionResponse_Failure_:
-			return nil, xerrors.WithStackTrace(xerrors.FromOperation(t.Failure))
-		case *Ydb_Coordination.SessionResponse_Ping:
-			err := stream.Send(&Ydb_Coordination.SessionRequest{
-				Request: &Ydb_Coordination.SessionRequest_Pong{
-					Pong: &Ydb_Coordination.SessionRequest_PingPong{
-						Opaque: t.Ping.GetOpaque(),
-					},
-				},
-			})
-			if err != nil {
-				return nil, xerrors.WithStackTrace(err)
-			}
-		default:
-			return nil, xerrors.WithStackTrace(fmt.Errorf("unexpected first message: %+v", msg))
-		}
-	}
-}
-
 // Create a new gRPC stream using an independent context.
-func (s *session) newStream(
+func (s *session) newStream( //nolint:funlen
 	streamCtx context.Context,
 	cancelStream context.CancelFunc,
 ) (Ydb_Coordination_V1.CoordinationService_SessionClient, error) {
@@ -230,8 +153,8 @@ func (s *session) newStream(
 				sessionClient Ydb_Coordination_V1.CoordinationService_SessionClient
 				err           error
 			)
-			onDone := trace.CoordinationOnNewSessionClient(s.trace, &streamCtx,
-				stack.FunctionID(""),
+			onDone := trace.CoordinationOnSessionNewStream(s.trace, &streamCtx,
+				stack.FunctionID("github.com/ydb-platform/ydb-go-sdk/v3/internal/coordination.(*session).newStream"),
 			)
 			sessionClient, err = s.client.Session(streamCtx)
 			onDone(err)
@@ -294,7 +217,7 @@ func (s *session) newStream(
 	}
 }
 
-func (s *session) mainLoop(ctx context.Context, path string, sessionStartedChan chan struct{}) {
+func (s *session) mainLoop(ctx context.Context, path string, sessionStartedChan chan struct{}) { //nolint:funlen
 	defer func() {
 		for _, f := range s.onClose {
 			f(s)
@@ -351,8 +274,6 @@ func (s *session) mainLoop(ctx context.Context, path string, sessionStartedChan
 			},
 		}
 
-		fmt.Printf("[SEND]: %+v (%T)\n", startSession.GetRequest(), startSession.GetRequest())
-
 		err = sessionClient.Send(&startSession)
 		if err != nil {
 			// Reconnect if a session cannot be started in this stream.
@@ -482,7 +403,7 @@ func (s *session) mainLoop(ctx context.Context, path string, sessionStartedChan
 	}
 }
 
-func (s *session) receiveLoop(
+func (s *session) receiveLoop( //nolint:funlen
 	wg *sync.WaitGroup,
 	sessionClient Ydb_Coordination_V1.CoordinationService_SessionClient,
 	cancelStream context.CancelFunc,
@@ -505,8 +426,6 @@ func (s *session) receiveLoop(
 		}
 		onDone(message, nil)
 
-		fmt.Printf("[RECV]: %+v (%T)\n", message.GetResponse(), message.GetResponse())
-
 		switch message.GetResponse().(type) {
 		case *Ydb_Coordination.SessionResponse_Failure_:
 			if message.GetFailure().GetStatus() == Ydb.StatusIds_SESSION_EXPIRED ||
@@ -589,8 +508,6 @@ func (s *session) sendLoop(
 			return
 		}
 
-		fmt.Printf("[SEND]: %+v (%T)\n", message.GetRequest(), message.GetRequest())
-
 		onSendDone := trace.CoordinationOnSessionSend(s.trace, message)
 		err = sessionClient.Send(message)
 		if err != nil {
@@ -861,7 +778,7 @@ func convertSemaphoreSession(
 	return &result
 }
 
-func (s *session) AcquireSemaphore(
+func (s *session) AcquireSemaphore( //nolint:funlen
 	ctx context.Context,
 	name string,
 	count uint64,

internal/coordination/session.go

@@ -169,10 +169,10 @@ func internalCoordination(
 				}
 			}
 		},
-		OnNewSessionClient: func(
-			info trace.CoordinationNewSessionClientStartInfo,
+		OnSessionNewStream: func(
+			info trace.CoordinationSessionNewStreamStartInfo,
 		) func(
-			info trace.CoordinationNewSessionClientDoneInfo,
+			info trace.CoordinationSessionNewStreamDoneInfo,
 		) {
 			if d.Details()&trace.CoordinationEvents == 0 {
 				return nil
@@ -181,7 +181,7 @@ func internalCoordination(
 			l.Log(ctx, "stream")
 			start := time.Now()
 
-			return func(info trace.CoordinationNewSessionClientDoneInfo) {
+			return func(info trace.CoordinationSessionNewStreamDoneInfo) {
 				l.Log(ctx, "done",
 					kv.Latency(start),
 					kv.Error(info.Error),

internal/coordination/session_test.go

@@ -30,7 +30,7 @@ func TestCoordinationSemaphore(t *testing.T) {
 	if err != nil {
 		t.Fatalf("failed to connect: %v", err)
 	}
-	defer db.Close(ctx) // cleanup resources
+	defer db.Close(ctx) // cleanup resourcess
 
 	const nodePath = "/local/coordination/node/test"