Skip to content

Commit b24c825

Browse files
KirillKurdyukovKirillKurdyukov
committed
feat: added first integration tests.yml
1 parent 253b08c commit b24c825

File tree

9 files changed

+201
-84
lines changed

9 files changed

+201
-84
lines changed

‎.github/workflows/tests.yml‎

Lines changed: 40 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -33,7 +33,6 @@ jobs:
3333
run: |
3434
cd src
3535
dotnet test --filter "Category=Unit" -f ${{ matrix.dotnet-target-framework }}
36-
3736
ado-net-tests:
3837
runs-on: ubuntu-22.04
3938
strategy:
@@ -74,7 +73,46 @@ jobs:
7473
docker cp ydb-local:/ydb_certs/ca.pem ~/
7574
cd src
7675
dotnet test --filter "(FullyQualifiedName~Ado) | (FullyQualifiedName~Dapper)" -l "console;verbosity=normal"
77-
76+
topic-tests:
77+
runs-on: ubuntu-22.04
78+
strategy:
79+
fail-fast: false
80+
matrix:
81+
ydb-version: [ 'trunk' ]
82+
dotnet-version: [ 6.0.x, 7.0.x ]
83+
include:
84+
- dotnet-version: 6.0.x
85+
dotnet-target-framework: net6.0
86+
- dotnet-version: 7.0.x
87+
dotnet-target-framework: net7.0
88+
services:
89+
ydb:
90+
image: cr.yandex/yc/yandex-docker-local-ydb:${{ matrix.ydb-version }}
91+
ports:
92+
- 2135:2135
93+
- 2136:2136
94+
- 8765:8765
95+
env:
96+
YDB_LOCAL_SURVIVE_RESTART: true
97+
YDB_USE_IN_MEMORY_PDISKS: true
98+
options: '--name ydb-local -h localhost'
99+
env:
100+
OS: ubuntu-22.04
101+
YDB_VERSION: ${{ matrix.ydb-version }}
102+
YDB_CONNECTION_STRING: grpc://localhost:2136/local
103+
YDB_CONNECTION_STRING_SECURE: grpcs://localhost:2135/local
104+
steps:
105+
- name: Checkout code
106+
uses: actions/checkout@v4
107+
- name: Install Dotnet
108+
uses: actions/setup-dotnet@v4
109+
with:
110+
dotnet-version: ${{ matrix.dotnet-version }}
111+
- name: Run ADO.NET tests
112+
run: |
113+
docker cp ydb-local:/ydb_certs/ca.pem ~/
114+
cd src
115+
dotnet test --filter "FullyQualifiedName~Topic" -l "console;verbosity=normal"
78116
core-integration-tests:
79117
runs-on: ubuntu-22.04
80118
strategy:

‎src/Ydb.Sdk/src/Services/Topic/Exceptions.cs‎

Lines changed: 13 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -1,20 +1,26 @@
11
namespace Ydb.Sdk.Services.Topic;
22

3-
public class YdbTopicException : Exception
3+
public class YdbWriterException : Exception
44
{
5-
protected YdbTopicException(string message) : base(message)
5+
public YdbWriterException(string message) : base(message)
66
{
7+
Status = new Status(StatusCode.Unspecified);
78
}
8-
}
99

10-
public class YdbWriterException : YdbTopicException
11-
{
12-
public YdbWriterException(string message) : base(message)
10+
public YdbWriterException(string message, Status status) : base(message + ": " + status)
1311
{
12+
Status = status;
1413
}
14+
15+
public YdbWriterException(string message, Driver.TransportException e) : base(message, e)
16+
{
17+
Status = e.Status;
18+
}
19+
20+
public Status Status { get; }
1521
}
1622

17-
public class YdbReaderException : YdbTopicException
23+
public class YdbReaderException : Exception
1824
{
1925
protected YdbReaderException(string message) : base(message)
2026
{

‎src/Ydb.Sdk/src/Services/Topic/TopicClient.cs‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -6,9 +6,9 @@ namespace Ydb.Sdk.Services.Topic;
66

77
public class TopicClient
88
{
9-
private readonly Driver _driver;
9+
private readonly IDriver _driver;
1010

11-
public TopicClient(Driver driver)
11+
public TopicClient(IDriver driver)
1212
{
1313
_driver = driver;
1414
}

‎src/Ydb.Sdk/src/Services/Topic/TopicSession.cs‎

Lines changed: 2 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -11,7 +11,6 @@ internal abstract class TopicSession<TFromClient, TFromServer> : IDisposable
1111
protected readonly string SessionId;
1212

1313
private int _isActive = 1;
14-
private bool _disposed;
1514

1615
protected TopicSession(BidirectionalStream<TFromClient, TFromServer> stream, ILogger logger,
1716
string sessionId, Func<Task> initialize)
@@ -32,28 +31,12 @@ protected async void ReconnectSession()
3231
}
3332

3433
Logger.LogInformation("WriterSession[{SessionId}] has been deactivated, starting to reconnect", SessionId);
35-
36-
while (!_disposed)
37-
{
38-
try
39-
{
40-
await _initialize();
41-
break;
42-
}
43-
catch (Exception e)
44-
{
45-
Logger.LogError(e, "Unable to reconnect the session due to the following error");
46-
}
47-
}
34+
35+
await _initialize();
4836
}
4937

5038
public void Dispose()
5139
{
52-
lock (this)
53-
{
54-
_disposed = true;
55-
}
56-
5740
Stream.Dispose();
5841
}
5942
}

‎src/Ydb.Sdk/src/Services/Topic/Writer/Writer.cs‎

Lines changed: 84 additions & 46 deletions
Original file line numberDiff line numberDiff line change
@@ -24,10 +24,10 @@ internal class Writer<TValue> : IWriter<TValue>
2424
private readonly ILogger<Writer<TValue>> _logger;
2525
private readonly ISerializer<TValue> _serializer;
2626
private readonly ConcurrentQueue<MessageSending> _toSendBuffer = new();
27+
private readonly CancellationTokenSource _disposeTokenSource = new();
2728

2829
private volatile TaskCompletionSource _taskWakeUpCompletionSource = new();
29-
private volatile IWriteSession _session = null!;
30-
private volatile bool _disposed;
30+
private volatile IWriteSession _session = new NotStartedWriterSession("Session not started!");
3131

3232
private int _limitBufferMaxSize;
3333

@@ -55,7 +55,7 @@ public async Task<WriteResult> WriteAsync(Message<TValue> message)
5555
var messageData = new MessageData
5656
{
5757
Data = ByteString.CopyFrom(data),
58-
CreatedAt = Timestamp.FromDateTime(message.Timestamp),
58+
CreatedAt = Timestamp.FromDateTime(message.Timestamp.ToUniversalTime()),
5959
UncompressedSize = data.Length
6060
};
6161

@@ -111,7 +111,7 @@ private async void StartWriteWorker()
111111
{
112112
await Initialize();
113113

114-
while (!_disposed)
114+
while (!_disposeTokenSource.Token.IsCancellationRequested)
115115
{
116116
await _taskWakeUpCompletionSource.Task;
117117
_taskWakeUpCompletionSource = new TaskCompletionSource();
@@ -127,76 +127,100 @@ private void WakeUpWorker()
127127

128128
private async Task Initialize()
129129
{
130-
_logger.LogInformation("Writer session initialization started. WriterConfig: {WriterConfig}", _config);
130+
try
131+
{
132+
_logger.LogInformation("Writer session initialization started. WriterConfig: {WriterConfig}", _config);
131133

132-
var stream = _driver.BidirectionalStreamCall(
133-
TopicService.StreamWriteMethod,
134-
GrpcRequestSettings.DefaultInstance
135-
);
134+
var stream = _driver.BidirectionalStreamCall(
135+
TopicService.StreamWriteMethod,
136+
GrpcRequestSettings.DefaultInstance
137+
);
136138

137-
var initRequest = new StreamWriteMessage.Types.InitRequest { Path = _config.TopicPath };
138-
if (_config.ProducerId != null)
139-
{
140-
initRequest.ProducerId = _config.ProducerId;
141-
}
139+
var initRequest = new StreamWriteMessage.Types.InitRequest { Path = _config.TopicPath };
140+
if (_config.ProducerId != null)
141+
{
142+
initRequest.ProducerId = _config.ProducerId;
143+
}
142144

143-
if (_config.MessageGroupId != null)
144-
{
145-
initRequest.MessageGroupId = _config.MessageGroupId;
146-
}
145+
if (_config.MessageGroupId != null)
146+
{
147+
initRequest.MessageGroupId = _config.MessageGroupId;
148+
}
147149

148-
_logger.LogDebug("Sending initialization request for the write stream: {InitRequest}", initRequest);
150+
_logger.LogDebug("Sending initialization request for the write stream: {InitRequest}", initRequest);
149151

150-
await stream.Write(new MessageFromClient { InitRequest = initRequest });
151-
if (!await stream.MoveNextAsync())
152-
{
153-
_session = new NotStartedWriterSession(
154-
$"Stream unexpectedly closed by YDB server. Current InitRequest: {initRequest}");
152+
await stream.Write(new MessageFromClient { InitRequest = initRequest });
153+
if (!await stream.MoveNextAsync())
154+
{
155+
_session = new NotStartedWriterSession(
156+
$"Stream unexpectedly closed by YDB server. Current InitRequest: {initRequest}");
155157

156-
_ = Task.Run(Initialize);
158+
_ = Task.Run(Initialize, _disposeTokenSource.Token);
157159

158-
return;
159-
}
160+
return;
161+
}
160162

161-
var receivedInitMessage = stream.Current;
163+
var receivedInitMessage = stream.Current;
162164

163-
var status = Status.FromProto(receivedInitMessage.Status, receivedInitMessage.Issues);
165+
var status = Status.FromProto(receivedInitMessage.Status, receivedInitMessage.Issues);
164166

165-
if (status.IsNotSuccess)
166-
{
167-
_session = new NotStartedWriterSession(status.ToString());
167+
if (status.IsNotSuccess)
168+
{
169+
_session = new NotStartedWriterSession("Initialization failed", status);
168170

169-
_ = Task.Run(Initialize);
171+
if (status.StatusCode != StatusCode.SchemeError)
172+
{
173+
_ = Task.Run(Initialize, _disposeTokenSource.Token);
174+
}
170175

171-
return;
172-
}
176+
return;
177+
}
173178

174-
var initResponse = receivedInitMessage.InitResponse;
179+
var initResponse = receivedInitMessage.InitResponse;
175180

176-
_logger.LogDebug("Received a response for the initialization request on the write stream: {InitResponse}",
177-
initResponse);
181+
_logger.LogDebug("Received a response for the initialization request on the writer stream: {InitResponse}",
182+
initResponse);
178183

179-
if (!initResponse.SupportedCodecs.Codecs.Contains((int)_config.Codec))
184+
if (initResponse.SupportedCodecs != null &&
185+
!initResponse.SupportedCodecs.Codecs.Contains((int)_config.Codec))
186+
{
187+
_logger.LogCritical("Topic[{TopicPath}] is not supported codec: {Codec}", _config.TopicPath,
188+
_config.Codec);
189+
190+
_session = new NotStartedWriterSession(
191+
$"Topic[{_config.TopicPath}] is not supported codec: {_config.Codec}");
192+
return;
193+
}
194+
195+
_session = new WriterSession(_config, stream, initResponse, Initialize, _logger);
196+
}
197+
catch (Driver.TransportException e)
180198
{
181-
_logger.LogCritical("Topic[{TopicPath}] is not supported codec: {Codec}", _config.TopicPath, _config.Codec);
199+
_logger.LogError(e, "Unable to connect the session");
182200

183201
_session = new NotStartedWriterSession(
184-
$"Topic[{_config.TopicPath}] is not supported codec: {_config.Codec}");
185-
return;
202+
new YdbWriterException("Transport error on creating write session", e));
186203
}
187-
188-
_session = new WriterSession(_config, stream, initResponse, Initialize, _logger);
189204
}
190205

191206
public void Dispose()
192207
{
193-
_disposed = true;
208+
try
209+
{
210+
_disposeTokenSource.Cancel();
211+
212+
_session.Dispose();
213+
}
214+
finally
215+
{
216+
_disposeTokenSource.Dispose();
217+
}
194218
}
195219
}
196220

197221
internal record MessageSending(MessageData MessageData, TaskCompletionSource<WriteResult> TaskCompletionSource);
198222

199-
internal interface IWriteSession
223+
internal interface IWriteSession : IDisposable
200224
{
201225
Task Write(ConcurrentQueue<MessageSending> toSendBuffer);
202226
}
@@ -210,6 +234,16 @@ public NotStartedWriterSession(string reasonExceptionMessage)
210234
_reasonException = new YdbWriterException(reasonExceptionMessage);
211235
}
212236

237+
public NotStartedWriterSession(string reasonExceptionMessage, Status status)
238+
{
239+
_reasonException = new YdbWriterException(reasonExceptionMessage, status);
240+
}
241+
242+
public NotStartedWriterSession(YdbWriterException reasonException)
243+
{
244+
_reasonException = reasonException;
245+
}
246+
213247
public Task Write(ConcurrentQueue<MessageSending> toSendBuffer)
214248
{
215249
foreach (var messageSending in toSendBuffer)
@@ -219,6 +253,10 @@ public Task Write(ConcurrentQueue<MessageSending> toSendBuffer)
219253

220254
return Task.CompletedTask;
221255
}
256+
257+
public void Dispose()
258+
{
259+
}
222260
}
223261

224262
// No thread safe

‎src/Ydb.Sdk/src/Services/Topic/Writer/WriterBuilder.cs‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -3,9 +3,9 @@ namespace Ydb.Sdk.Services.Topic.Writer;
33
public class WriterBuilder<TValue>
44
{
55
private readonly WriterConfig _config;
6-
private readonly Driver _driver;
6+
private readonly IDriver _driver;
77

8-
public WriterBuilder(Driver driver, WriterConfig config)
8+
public WriterBuilder(IDriver driver, WriterConfig config)
99
{
1010
_driver = driver;
1111
_config = config;

‎src/Ydb.Sdk/tests/Fixture/DriverFixture.cs‎

Lines changed: 7 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -2,21 +2,23 @@
22

33
namespace Ydb.Sdk.Tests.Fixture;
44

5-
public abstract class DriverFixture : IAsyncLifetime
5+
public class DriverFixture : IAsyncLifetime
66
{
7-
protected readonly Driver Driver;
7+
public Driver Driver { get; }
88

9-
protected DriverFixture(DriverConfig? driverConfig = null)
9+
public DriverFixture()
1010
{
11-
driverConfig ??= new DriverConfig(
11+
var driverConfig = new DriverConfig(
1212
endpoint: "grpc://localhost:2136",
1313
database: "/local"
1414
);
1515

1616
Driver = new Driver(driverConfig, Utils.GetLoggerFactory());
1717
}
1818

19-
protected abstract void ClientDispose();
19+
protected virtual void ClientDispose()
20+
{
21+
}
2022

2123
public Task InitializeAsync()
2224
{

0 commit comments

Comments
 (0)