Browse Source

InputStream reads should wait until data is available.

tags/1.0-rc
RogueException 8 years ago
parent
commit
909127d330
3 changed files with 21 additions and 17 deletions
  1. +2
    -2
      src/Discord.Net.Core/Audio/AudioInStream.cs
  2. +1
    -0
      src/Discord.Net.Core/Audio/IAudioClient.cs
  3. +18
    -15
      src/Discord.Net.WebSocket/Audio/Streams/InputStream.cs

+ 2
- 2
src/Discord.Net.Core/Audio/AudioInStream.cs View File

@@ -11,9 +11,9 @@ namespace Discord.Audio
public override bool CanSeek => false; public override bool CanSeek => false;
public override bool CanWrite => true; public override bool CanWrite => true;


public abstract Task<RTPFrame?> ReadFrameAsync(CancellationToken cancelToken);
public abstract Task<RTPFrame> ReadFrameAsync(CancellationToken cancelToken);


public RTPFrame? ReadFrame()
public RTPFrame ReadFrame()
{ {
return ReadFrameAsync(CancellationToken.None).GetAwaiter().GetResult(); return ReadFrameAsync(CancellationToken.None).GetAwaiter().GetResult();
} }


+ 1
- 0
src/Discord.Net.Core/Audio/IAudioClient.cs View File

@@ -10,6 +10,7 @@ namespace Discord.Audio
event Func<int, int, Task> LatencyUpdated; event Func<int, int, Task> LatencyUpdated;
event Func<ulong, AudioInStream, Task> StreamCreated; event Func<ulong, AudioInStream, Task> StreamCreated;
event Func<ulong, Task> StreamDestroyed; event Func<ulong, Task> StreamDestroyed;
event Func<ulong, bool, Task> SpeakingUpdated;


/// <summary> Gets the current connection state of this client. </summary> /// <summary> Gets the current connection state of this client. </summary>
ConnectionState ConnectionState { get; } ConnectionState ConnectionState { get; }


+ 18
- 15
src/Discord.Net.WebSocket/Audio/Streams/InputStream.cs View File

@@ -8,7 +8,10 @@ namespace Discord.Audio.Streams
///<summary> Reads the payload from an RTP frame </summary> ///<summary> Reads the payload from an RTP frame </summary>
public class InputStream : AudioInStream public class InputStream : AudioInStream
{ {
private const int MaxFrames = 100;

private ConcurrentQueue<RTPFrame> _frames; private ConcurrentQueue<RTPFrame> _frames;
private SemaphoreSlim _signal;
private ushort _nextSeq; private ushort _nextSeq;
private uint _nextTimestamp; private uint _nextTimestamp;
private bool _hasHeader; private bool _hasHeader;
@@ -21,28 +24,27 @@ namespace Discord.Audio.Streams
public InputStream() public InputStream()
{ {
_frames = new ConcurrentQueue<RTPFrame>(); _frames = new ConcurrentQueue<RTPFrame>();
_signal = new SemaphoreSlim(0, MaxFrames);
} }


public override Task<RTPFrame?> ReadFrameAsync(CancellationToken cancelToken)
public override async Task<RTPFrame> ReadFrameAsync(CancellationToken cancelToken)
{ {
cancelToken.ThrowIfCancellationRequested(); cancelToken.ThrowIfCancellationRequested();


if (_frames.TryDequeue(out var frame))
return Task.FromResult<RTPFrame?>(frame);
return Task.FromResult<RTPFrame?>(null);
RTPFrame frame;
await _signal.WaitAsync(cancelToken).ConfigureAwait(false);
_frames.TryDequeue(out frame);
return frame;
} }
public override Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancelToken)
public override async Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancelToken)
{ {
cancelToken.ThrowIfCancellationRequested(); cancelToken.ThrowIfCancellationRequested();
if (_frames.TryDequeue(out var frame))
{
if (count < frame.Payload.Length)
throw new InvalidOperationException("Buffer is too small.");
Buffer.BlockCopy(frame.Payload, 0, buffer, offset, frame.Payload.Length);
return Task.FromResult(frame.Payload.Length);
}
return Task.FromResult(0);

var frame = await ReadFrameAsync(cancelToken).ConfigureAwait(false);
if (count < frame.Payload.Length)
throw new InvalidOperationException("Buffer is too small.");
Buffer.BlockCopy(frame.Payload, 0, buffer, offset, frame.Payload.Length);
return frame.Payload.Length;
} }


public void WriteHeader(ushort seq, uint timestamp) public void WriteHeader(ushort seq, uint timestamp)
@@ -57,7 +59,7 @@ namespace Discord.Audio.Streams
{ {
cancelToken.ThrowIfCancellationRequested(); cancelToken.ThrowIfCancellationRequested();


if (_frames.Count > 100) //1-2 seconds
if (_frames.Count > MaxFrames) //1-2 seconds
{ {
_hasHeader = false; _hasHeader = false;
return Task.Delay(0); //Buffer overloaded return Task.Delay(0); //Buffer overloaded
@@ -72,6 +74,7 @@ namespace Discord.Audio.Streams
timestamp: _nextTimestamp, timestamp: _nextTimestamp,
payload: payload payload: payload
)); ));
_signal.Release();
_hasHeader = false; _hasHeader = false;
return Task.Delay(0); return Task.Delay(0);
} }


Loading…
Cancel
Save