Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merge hotfix/fixReceiveAsyncAndExamples into develop #13

Merged
merged 2 commits into from
Jun 1, 2024
Merged

Merge hotfix/fixReceiveAsyncAndExamples into develop #13

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
a329507
fix receiveAsync loop and examples
HandyS11 Jun 1, 2024
9850cb0
remove id
HandyS11 Jun 1, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 4 additions & 8 deletions RustPlusApi/Examples/GetAlarmInfo/Program.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,13 +7,9 @@
var rustPlus = new RustPlus(Ip, Port, PlayerId, PlayerToken);
const uint entityId = 0;

rustPlus.Connected += async (_, _) =>
{
var message = await rustPlus.GetAlarmInfoAsync(entityId);
await rustPlus.ConnectAsync();

Console.WriteLine($"Infos:\n{JsonConvert.SerializeObject(message, JsonSettings)}");
var message = await rustPlus.GetAlarmInfoAsync(entityId);
Console.WriteLine($"Infos:\n{JsonConvert.SerializeObject(message, JsonSettings)}");

rustPlus.Dispose();
};

await rustPlus.ConnectAsync();
await rustPlus.DisconnectAsync();
5 changes: 4 additions & 1 deletion RustPlusApi/Examples/Legacy/GetClanChatChangesLegacy/Program.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,4 +17,7 @@

// This method is not fully integrated in Rust, so it will not work until the Clan update is released.
var message = await rustPlus.GetClanChatLegacyAsync();
Console.WriteLine($"Infos:\n{JsonConvert.SerializeObject(message, JsonSettings)}");
Console.WriteLine($"Infos:\n{JsonConvert.SerializeObject(message, JsonSettings)}");

Console.ReadLine();
await rustPlus.DisconnectAsync();
7 changes: 5 additions & 2 deletions RustPlusApi/Examples/Legacy/GetEntityChangesLegacy/Program.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@
using static __Constants.ExamplesConst;

var rustPlus = new RustPlusLegacy(Ip, Port, PlayerId, PlayerToken);
const uint entityId = 3716008;
const uint entityId = 0;

rustPlus.NotificationReceived += (_, message) =>
{
Expand All @@ -17,4 +17,7 @@
await rustPlus.ConnectAsync();

var message = await rustPlus.GetEntityInfoLegacyAsync(entityId);
Console.WriteLine($"Infos:\n{JsonConvert.SerializeObject(message, JsonSettings)}");
Console.WriteLine($"Infos:\n{JsonConvert.SerializeObject(message, JsonSettings)}");

Console.ReadLine();
await rustPlus.DisconnectAsync();
5 changes: 4 additions & 1 deletion RustPlusApi/Examples/Legacy/GetTeamChatChangesLegacy/Program.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,4 +16,7 @@
await rustPlus.ConnectAsync();

var message = await rustPlus.GetTeamChatLegacyAsync();
Console.WriteLine($"Infos:\n{JsonConvert.SerializeObject(message, JsonSettings)}");
Console.WriteLine($"Infos:\n{JsonConvert.SerializeObject(message, JsonSettings)}");

Console.ReadLine();
await rustPlus.DisconnectAsync();
2 changes: 1 addition & 1 deletion RustPlusApi/Examples/Legacy/__Events/Program.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,7 @@
var request = new AppRequest
{
GetTime = new AppEmpty()
};
};
await rustPlus.SendRequestAsync(request);

await rustPlus.DisconnectAsync();
Expand Down
37 changes: 24 additions & 13 deletions RustPlusApi/RustPlusApi/RustPlusSocket.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,7 +30,7 @@ public abstract class RustPlusSocket(string server, int port, ulong playerId, in
private readonly ConcurrentQueue<TaskCompletionSource<AppMessage>> _responseQueue = new();

private readonly CancellationTokenSource _cancellationTokenSource = new();
private CancellationToken _cancellationToken;
private CancellationToken _cancellationToken => _cancellationTokenSource.Token;

public event EventHandler? Connecting;
public event EventHandler? Connected;
Expand Down Expand Up @@ -66,10 +66,8 @@ public async Task ConnectAsync()
{
await _webSocket.ConnectAsync(uri, CancellationToken.None);

_ = Task.Run(ReceiveAsync, _cancellationToken);
_ = Task.Run(ProcessSendQueueAsync, _cancellationToken);

_cancellationToken = _cancellationTokenSource.Token;
_ = Task.Run(ReceiveAsync, CancellationToken.None);
_ = Task.Run(ProcessSendQueueAsync, CancellationToken.None);

Connected?.Invoke(this, EventArgs.Empty);
}
Expand Down Expand Up @@ -110,14 +108,14 @@ public async Task<AppMessage> SendRequestAsync(AppRequest request)
/// <returns>A task representing the asynchronous operation.</returns>
private async Task ProcessSendQueueAsync()
{
while (IsConnected())
while (IsConnected() && !_cancellationToken.IsCancellationRequested)
{
if (_sendQueue.TryDequeue(out var request))
{
var buffer = request.ToByteArray();
await _webSocket!.SendAsync(new ArraySegment<byte>(buffer), WebSocketMessageType.Binary, true, _cancellationToken);
await _webSocket!.SendAsync(new ArraySegment<byte>(buffer), WebSocketMessageType.Binary, true, CancellationToken.None);
}
await Task.Delay(100, _cancellationToken);
await Task.Delay(100, CancellationToken.None);
}
}

Expand All @@ -130,37 +128,49 @@ private async Task ReceiveAsync()
const int bufferSize = 1024;
var buffer = new byte[bufferSize];

while (IsConnected())
Debug.WriteLine("Receiving data from the Rust+ server...");

while (IsConnected() && !_cancellationToken.IsCancellationRequested)
{
Debug.WriteLine("Waiting for data...");
try
{
var receiveBuffer = new List<byte>();
WebSocketReceiveResult result;

do
{
result = await _webSocket!.ReceiveAsync(new ArraySegment<byte>(buffer), _cancellationToken);
result = await _webSocket!.ReceiveAsync(new ArraySegment<byte>(buffer), CancellationToken.None);
receiveBuffer.AddRange(buffer.Take(result.Count));
} while (!result.EndOfMessage);

var messageData = receiveBuffer.ToArray();
var message = AppMessage.Parser.ParseFrom(messageData);

Debug.WriteLine($"Received message:\n{message}");
MessageReceived?.Invoke(this, message);

if (message.Response?.Seq == 0)
if (message.Broadcast is not null)
{
Debug.WriteLine($"Received notification:\n{message}");
NotificationReceived?.Invoke(this, message);
ParseNotification(message.Broadcast);
}
else
{
Debug.WriteLine($"Received response:\n{message}");
ResponseReceived?.Invoke(this, message);
}

if (_responseQueue.TryDequeue(out var tcs))
tcs.SetResult(message);
_ = Task.Run(() =>
{
if (_responseQueue.TryDequeue(out var tcs))
tcs.SetResult(message);
});
}
catch (OperationCanceledException)
{
Debug.WriteLine("Operation canceled.");
break;
}
catch (WebSocketException ex)
Expand All @@ -174,6 +184,7 @@ private async Task ReceiveAsync()
ErrorOccurred?.Invoke(this, ex);
}
}
Debug.WriteLine("Receive loop exited.");
}

/// <summary>
Expand Down
Loading