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
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
|
using Microsoft.AspNet.SignalR.Client;
using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using Tango.Core;
namespace Tango.Transport.Adapters
{
/// <summary>
/// Represents an adapter for communicating via SignalR protocol.
/// </summary>
/// <seealso cref="Tango.Transport.TransportAdapterBase" />
public class SignalRTransportAdapter : TransportAdapterBase
{
private IHubProxy _proxy;
private HubConnection _connection;
private Thread _pushThread;
private ProducerConsumerQueue<byte[]> _pushQueue;
/// <summary>
/// Gets or sets the URL of the SignalR service.
/// </summary>
public String Url { get; set; }
/// <summary>
/// Gets or sets the SignalR hub name.
/// </summary>
public String Hub { get; set; }
/// <summary>
/// Gets or sets the serial number of the remote machine (Use onlt for <see cref="SignalRTransportAdapterMode.CreateSession"/>) mode.
/// </summary>
public String SerialNumber { get; set; }
/// <summary>
/// Gets or sets the remote session identifier.
/// </summary>
public String SessionID { get; private set; }
/// <summary>
/// Gets or sets the interval of write operation.
/// Unlike other adapters, the SignalR adapter accumulates multiple write operations into one chunk.
/// </summary>
public TimeSpan WriteInterval { get; set; }
/// <summary>
/// Gets or sets the adapter mode.
/// </summary>
public SignalRTransportAdapterMode Mode { get; set; }
/// <summary>
/// Gets or sets a value indicating whether to enable compression/decompression of data.
/// </summary>
public bool EnableCompression { get; set; }
/// <summary>
/// Initializes a new instance of the <see cref="SignalRTransportAdapter"/> class.
/// </summary>
public SignalRTransportAdapter() : base()
{
WriteInterval = TimeSpan.FromSeconds(1);
ComponentName = $"SignalR Adapter {_component_counter++}";
}
/// <summary>
/// Initializes a new instance of the <see cref="SignalRTransportAdapter"/> class.
/// </summary>
/// <param name="url">The service address.</param>
/// <param name="hub">The hub name.</param>
/// <param name="mode">The adapter mode.</param>
/// <param name="serialNumber">The machine serial number (when creating session).</param>
/// <param name="sessionID">The session identifier (when joining session).</param>
public SignalRTransportAdapter(String url, String hub, SignalRTransportAdapterMode mode, String serialNumber = null, String sessionID = null) : this()
{
Url = url;
Hub = hub;
Mode = mode;
SerialNumber = serialNumber;
SessionID = sessionID;
Address = sessionID;
}
/// <summary>
/// Writes the specified data to the stream.
/// </summary>
/// <param name="data">The data.</param>
public override void Write(byte[] data)
{
TotalBytesSent += data.Length;
_totalBytes += data.Length;
AppendTransferRateBytes(data.Length);
_pushQueue.BlockEnqueue(data);
}
/// <summary>
/// Connects the transport component.
/// </summary>
/// <returns></returns>
public override Task Connect()
{
if (State != TransportComponentState.Connected)
{
LogManager.Log("Connecting SignalR adapter...");
bool completed = false;
TaskCompletionSource<object> completionSource = new TaskCompletionSource<object>();
_connection = new HubConnection(Url);
_proxy = _connection.CreateHubProxy(Hub);
Core.Threading.TimeoutTask.StartNew(() =>
{
if (!completed)
{
completed = true;
completionSource.SetException(new TimeoutException("Could not connect the SignalR adapter."));
}
}, TimeSpan.FromSeconds(5));
if (Mode == SignalRTransportAdapterMode.CreateSession)
{
_proxy.On("SessionCreated", () =>
{
if (!completed)
{
LogManager.Log($"SignalR adapter session created ({SessionID})...");
LogManager.Log("SingalR adapter connected.");
completed = true;
State = TransportComponentState.Connected;
_pushThread = new Thread(PushThreadMethod);
_pushThread.IsBackground = true;
_pushThread.Name = $"{ComponentName} Push Thread";
_pushQueue = new ProducerConsumerQueue<byte[]>();
_pushThread.Start();
completionSource.SetResult(true);
}
});
}
_connection.StateChanged += async (x) =>
{
if (x.NewState == ConnectionState.Connected)
{
if (Mode == SignalRTransportAdapterMode.CreateSession)
{
LogManager.Log("Creating SignalR adapter Session...");
SessionID = await _proxy.Invoke<String>("CreateSession", SerialNumber);
Address = SessionID;
}
else
{
LogManager.Log($"Joining SignalR adapter session ({SessionID})...");
await _proxy.Invoke("JoinSession", SessionID);
LogManager.Log("SingalR adapter connected.");
}
if (Mode == SignalRTransportAdapterMode.JoinSession)
{
if (!completed)
{
completed = true;
State = TransportComponentState.Connected;
_pushThread.Start();
completionSource.SetResult(true);
}
}
}
};
_proxy.On<List<byte[]>>("DataAvailable", (dataCollection) => { OnDataAvailable(dataCollection); });
_connection.Start();
return completionSource.Task;
}
return Task.FromResult(true);
}
/// <summary>
/// Disconnects the transport component.
/// </summary>
/// <returns></returns>
public override Task Disconnect()
{
return Task.Factory.StartNew(() =>
{
if (State == TransportComponentState.Connected)
{
LogManager.Log("Disconnecting SignalR adapter...");
Core.Threading.TimeoutTask.StartNew(() =>
{
try
{
_connection.Stop();
_connection.Dispose();
try
{
_pushThread.Abort();
}
catch { }
}
catch (Exception ex)
{
LogManager.Log(ex, "Error disposing SignalR adapter connection.");
}
}, TimeSpan.FromSeconds(5));
State = TransportComponentState.Disconnected;
LogManager.Log("SignalR adapter disconnected.");
}
});
}
/// <summary>
/// Handles the write operation.
/// </summary>
private void PushThreadMethod()
{
try
{
while (State == TransportComponentState.Connected)
{
List<byte[]> dataCollection = new List<byte[]>();
var data = _pushQueue.BlockDequeue();
var first = true;
while (_pushQueue.Count > 0 || first)
{
if (!first)
{
data = _pushQueue.BlockDequeue();
}
else
{
first = false;
}
if (EnableCompression)
{
var compressed = Compression.SevenZipHelper.Compress(data);
dataCollection.Add(compressed);
}
else
{
dataCollection.Add(data);
}
}
if (dataCollection.Count > 0)
{
try
{
_proxy.Invoke("Write", dataCollection).GetAwaiter().GetResult();
}
catch (Exception ex)
{
OnFailed(ex);
return;
}
}
Thread.Sleep(WriteInterval);
}
}
catch (ThreadAbortException) { }
}
/// <summary>
/// Called when new data is available.
/// </summary>
/// <param name="dataCollection">The data collection.</param>
private void OnDataAvailable(List<byte[]> dataCollection)
{
foreach (var data in dataCollection)
{
if (EnableCompression)
{
OnDataAvailable(Compression.SevenZipHelper.Decompress(data));
}
else
{
OnDataAvailable(data);
}
}
}
}
}
|
