Repository navigation
Expand file tree
/
Copy pathVirtualConnection.cs
More file actions
154 lines (134 loc) · 5.58 KB
/
Copy pathVirtualConnection.cs
File metadata and controls
154 lines (134 loc) · 5.58 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
using System;
using System.IO;
using System.Threading;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;
namespace Microsoft.AspNetCore.NodeServices.Sockets.VirtualConnections
{
/// <summary>
/// A virtual read/write connection, typically to a remote process. Multiple virtual connections can be
/// multiplexed over a single physical connection (e.g., a named pipe, domain socket, or TCP socket).
/// </summary>
internal class VirtualConnection : Stream
{
#if NET451
private readonly static Task CompletedTask = Task.FromResult((object)null);
#else
private readonly static Task CompletedTask = Task.CompletedTask;
#endif
private VirtualConnectionClient _host;
private readonly BufferBlock<byte[]> _receivedDataQueue = new BufferBlock<byte[]>();
private ArraySegment<byte> _receivedDataNotYetUsed;
private bool _wasClosedByRemote;
private bool _isDisposed;
public VirtualConnection(long id, VirtualConnectionClient host)
{
Id = id;
_host = host;
}
public long Id { get; }
public override bool CanRead { get { return true; } }
public override bool CanSeek { get { return false; } }
public override bool CanWrite { get { return true; } }
public override long Length
{
get { throw new NotImplementedException(); }
}
public override long Position
{
get { throw new NotImplementedException(); }
set { throw new NotImplementedException(); }
}
public override void Flush()
{
// We're auto-flushing, so this is a no-op.
}
public override async Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
{
if (_wasClosedByRemote)
{
return 0;
}
var bytesRead = 0;
while (true)
{
// Pull as many applicable bytes as we can out of receivedDataNotYetUsed, then update its offset/length
int bytesToExtract = Math.Min(count - bytesRead, _receivedDataNotYetUsed.Count);
if (bytesToExtract > 0)
{
Buffer.BlockCopy(_receivedDataNotYetUsed.Array, _receivedDataNotYetUsed.Offset, buffer, bytesRead, bytesToExtract);
_receivedDataNotYetUsed = new ArraySegment<byte>(_receivedDataNotYetUsed.Array, _receivedDataNotYetUsed.Offset + bytesToExtract, _receivedDataNotYetUsed.Count - bytesToExtract);
bytesRead += bytesToExtract;
}
// If we've completely filled the output buffer, we're done
if (bytesRead == count)
{
return bytesRead;
}
// We haven't yet filled the output buffer, so we must have exhausted receivedDataNotYetUsed instead.
// We want to get the next block of data from the underlying queue.
byte[] nextReceivedBlock;
if (bytesRead > 0)
{
if (!_receivedDataQueue.TryReceive(null, out nextReceivedBlock))
{
// No more data is available synchronously, and we already have some data, so we can stop now
return bytesRead;
}
}
else
{
// Since we don't yet have anything, wait for the underlying source
nextReceivedBlock = await _receivedDataQueue.ReceiveAsync(cancellationToken);
}
if (nextReceivedBlock.Length == 0)
{
// A zero-length block signals that the remote regards this virtual connection as closed
_wasClosedByRemote = true;
return bytesRead;
}
else
{
// We got some more data, so can continue trying to fill the output buffer
_receivedDataNotYetUsed = new ArraySegment<byte>(nextReceivedBlock, 0, nextReceivedBlock.Length);
}
}
}
public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
{
if (_wasClosedByRemote)
{
throw new InvalidOperationException("The connection was already closed by the remote party");
}
return count > 0 ? _host.WriteAsync(Id, buffer, offset, count, cancellationToken) : CompletedTask;
}
public override int Read(byte[] buffer, int offset, int count)
{
return ReadAsync(buffer, offset, count, CancellationToken.None).Result;
}
public override long Seek(long offset, SeekOrigin origin)
{
throw new NotImplementedException();
}
public override void SetLength(long value)
{
throw new NotImplementedException();
}
public override void Write(byte[] buffer, int offset, int count)
{
WriteAsync(buffer, offset, count, CancellationToken.None).Wait();
}
protected override void Dispose(bool disposing)
{
if (disposing && !_isDisposed)
{
_isDisposed = true;
_host.CloseInnerStream(Id, _wasClosedByRemote);
}
}
public async Task AddDataToQueue(byte[] data)
{
await _receivedDataQueue.SendAsync(data);
}
}
}