Nikita Tsukanov
9 years ago
16 changed files with 691 additions and 47 deletions
@ -0,0 +1,79 @@ |
|||
using System; |
|||
using System.Runtime.InteropServices; |
|||
using Avalonia.Input; |
|||
using Avalonia.Media; |
|||
using Avalonia.Media.Imaging; |
|||
using Avalonia.Remote.Protocol; |
|||
using Avalonia.Remote.Protocol.Viewport; |
|||
using Avalonia.Threading; |
|||
using PixelFormat = Avalonia.Platform.PixelFormat; |
|||
|
|||
namespace Avalonia.Controls.Remote |
|||
{ |
|||
public class RemoteWidget : Control |
|||
{ |
|||
private readonly IAvaloniaRemoteTransportConnection _connection; |
|||
private FrameMessage _lastFrame; |
|||
private WritableBitmap _bitmap; |
|||
public RemoteWidget(IAvaloniaRemoteTransportConnection connection) |
|||
{ |
|||
_connection = connection; |
|||
_connection.OnMessage += msg => Dispatcher.UIThread.InvokeAsync(() => OnMessage(msg)); |
|||
_connection.Send(new ClientSupportedPixelFormatsMessage |
|||
{ |
|||
Formats = new[] |
|||
{ |
|||
Avalonia.Remote.Protocol.Viewport.PixelFormat.Bgra8888, |
|||
Avalonia.Remote.Protocol.Viewport.PixelFormat.Rgba8888, |
|||
} |
|||
}); |
|||
} |
|||
|
|||
private void OnMessage(object msg) |
|||
{ |
|||
if (msg is FrameMessage frame) |
|||
{ |
|||
_connection.Send(new FrameReceivedMessage |
|||
{ |
|||
SequenceId = frame.SequenceId |
|||
}); |
|||
_lastFrame = frame; |
|||
InvalidateVisual(); |
|||
} |
|||
|
|||
} |
|||
|
|||
protected override void ArrangeCore(Rect finalRect) |
|||
{ |
|||
_connection.Send(new ClientViewportAllocatedMessage |
|||
{ |
|||
Width = finalRect.Width, |
|||
Height = finalRect.Height, |
|||
DpiX = 96, |
|||
DpiY = 96 //TODO: Somehow detect the actual DPI
|
|||
}); |
|||
base.ArrangeCore(finalRect); |
|||
} |
|||
|
|||
public override void Render(DrawingContext context) |
|||
{ |
|||
if (_lastFrame != null) |
|||
{ |
|||
var fmt = (PixelFormat) _lastFrame.Format; |
|||
if (_bitmap == null || _bitmap.PixelWidth != _lastFrame.Width || |
|||
_bitmap.PixelHeight != _lastFrame.Height) |
|||
_bitmap = new WritableBitmap(_lastFrame.Width, _lastFrame.Height, fmt); |
|||
using (var l = _bitmap.Lock()) |
|||
{ |
|||
var lineLen = (fmt == PixelFormat.Rgb565 ? 2 : 4) * _lastFrame.Width; |
|||
for (var y = 0; y < _lastFrame.Height; y++) |
|||
Marshal.Copy(_lastFrame.Data, y * _lastFrame.Stride, |
|||
new IntPtr(l.Address.ToInt64() + l.RowBytes * y), lineLen); |
|||
} |
|||
context.DrawImage(_bitmap, 1, new Rect(0, 0, _bitmap.PixelWidth, _bitmap.PixelHeight), |
|||
new Rect(Bounds.Size)); |
|||
} |
|||
base.Render(context); |
|||
} |
|||
} |
|||
} |
|||
@ -1,9 +1,10 @@ |
|||
<Project Sdk="Microsoft.NET.Sdk"> |
|||
<PropertyGroup> |
|||
<TargetFramework>netstandard1.3</TargetFramework> |
|||
<DefineConstants>AVALONIA_REMOTE_PROTOCOL;$(DefineConstants)</DefineConstants> |
|||
</PropertyGroup> |
|||
<ItemGroup> |
|||
<Compile Include="..\Avalonia.Input\Key.cs" /> |
|||
</ItemGroup> |
|||
</Project> |
|||
<Project Sdk="Microsoft.NET.Sdk"> |
|||
<PropertyGroup> |
|||
<TargetFramework>netstandard1.3</TargetFramework> |
|||
<DefineConstants>AVALONIA_REMOTE_PROTOCOL;$(DefineConstants)</DefineConstants> |
|||
</PropertyGroup> |
|||
<ItemGroup> |
|||
<PackageReference Include="Newtonsoft.Json" Version="9.0.1" /> |
|||
<Compile Include="..\Avalonia.Input\Key.cs" /> |
|||
</ItemGroup> |
|||
</Project> |
|||
@ -1,7 +1,160 @@ |
|||
namespace Avalonia.Remote.Protocol |
|||
using System; |
|||
using System.Collections.Generic; |
|||
using System.IO; |
|||
using System.Threading; |
|||
using Newtonsoft.Json; |
|||
using System.Threading.Tasks; |
|||
using Newtonsoft.Json.Bson; |
|||
|
|||
namespace Avalonia.Remote.Protocol |
|||
{ |
|||
public class BsonStreamTransport |
|||
public class BsonStreamTransportConnection : IAvaloniaRemoteTransportConnection |
|||
{ |
|||
private readonly IMessageTypeResolver _resolver; |
|||
private readonly Stream _inputStream; |
|||
private readonly Stream _outputStream; |
|||
private readonly Action _disposeCallback; |
|||
private readonly CancellationToken _cancel; |
|||
private readonly CancellationTokenSource _cancelSource; |
|||
private readonly MemoryStream _outputBlock = new MemoryStream(); |
|||
private readonly object _lock = new object(); |
|||
private bool _writeOperationPending; |
|||
private bool _readingAlreadyStarted; |
|||
private bool _writerIsBroken; |
|||
static readonly JsonSerializer Serializer = new JsonSerializer(); |
|||
private static readonly byte[] ZeroLength = new byte[4]; |
|||
|
|||
public BsonStreamTransportConnection(IMessageTypeResolver resolver, Stream inputStream, Stream outputStream, Action disposeCallback) |
|||
{ |
|||
_resolver = resolver; |
|||
_inputStream = inputStream; |
|||
_outputStream = outputStream; |
|||
_disposeCallback = disposeCallback; |
|||
_cancelSource = new CancellationTokenSource(); |
|||
_cancel = _cancelSource.Token; |
|||
} |
|||
|
|||
public void Dispose() |
|||
{ |
|||
_cancelSource.Cancel(); |
|||
_disposeCallback?.Invoke(); |
|||
} |
|||
|
|||
public void StartReading() |
|||
{ |
|||
lock (_lock) |
|||
{ |
|||
if(_readingAlreadyStarted) |
|||
throw new InvalidOperationException("Reading has already started"); |
|||
_readingAlreadyStarted = true; |
|||
Task.Run(Reader, _cancel); |
|||
} |
|||
} |
|||
|
|||
async Task ReadExact(byte[] buffer) |
|||
{ |
|||
int read = 0; |
|||
while (read != buffer.Length) |
|||
{ |
|||
var readNow = await _inputStream.ReadAsync(buffer, read, buffer.Length - read, _cancel) |
|||
.ConfigureAwait(false); |
|||
if (readNow == 0) |
|||
throw new EndOfStreamException(); |
|||
read += readNow; |
|||
} |
|||
} |
|||
|
|||
async Task Reader() |
|||
{ |
|||
Task.Yield(); |
|||
try |
|||
{ |
|||
while (true) |
|||
{ |
|||
var infoBlock = new byte[20]; |
|||
await ReadExact(infoBlock).ConfigureAwait(false); |
|||
var length = BitConverter.ToInt32(infoBlock, 0); |
|||
var guidBytes = new byte[16]; |
|||
Buffer.BlockCopy(infoBlock, 4, guidBytes, 0, 16); |
|||
var guid = new Guid(guidBytes); |
|||
var buffer = new byte[length]; |
|||
await ReadExact(buffer).ConfigureAwait(false); |
|||
if (Environment.GetEnvironmentVariable("WTF") == "WTF") |
|||
{ |
|||
|
|||
using (var f = System.IO.File.Create("/tmp/wtf2.bin")) |
|||
{ |
|||
f.Write(infoBlock, 0, infoBlock.Length); |
|||
f.Write(buffer, 0, buffer.Length); |
|||
} |
|||
} |
|||
var message = Serializer.Deserialize(new BsonReader(new MemoryStream(buffer)), _resolver.GetByGuid(guid)); |
|||
OnMessage?.Invoke(message); |
|||
} |
|||
} |
|||
catch (Exception e) |
|||
{ |
|||
FireException(e); |
|||
} |
|||
} |
|||
|
|||
|
|||
public async Task Send(object data) |
|||
{ |
|||
lock (_lock) |
|||
{ |
|||
if(_writerIsBroken) //Ignore further calls, since there is no point of writing to "broken" stream
|
|||
return; |
|||
if (_writeOperationPending) |
|||
throw new InvalidOperationException("Previous send operation was not finished"); |
|||
_writeOperationPending = true; |
|||
} |
|||
try |
|||
{ |
|||
var guid = _resolver.GetGuid(data.GetType()).ToByteArray(); |
|||
_outputBlock.Seek(0, SeekOrigin.Begin); |
|||
_outputBlock.SetLength(0); |
|||
_outputBlock.Write(ZeroLength, 0, 4); |
|||
_outputBlock.Write(guid, 0, guid.Length); |
|||
var writer = new BsonWriter(_outputBlock); |
|||
Serializer.Serialize(writer, data); |
|||
_outputBlock.Seek(0, SeekOrigin.Begin); |
|||
var length = BitConverter.GetBytes((int)_outputBlock.Length - 20); |
|||
_outputBlock.Write(length, 0, length.Length); |
|||
_outputBlock.Seek(0, SeekOrigin.Begin); |
|||
|
|||
try |
|||
{ |
|||
await _outputBlock.CopyToAsync(_outputStream, 0x1000, _cancel).ConfigureAwait(false); |
|||
} |
|||
catch (Exception e) //We are only catching "network"-related exceptions here
|
|||
{ |
|||
lock (_lock) |
|||
{ |
|||
_writerIsBroken = true; |
|||
} |
|||
FireException(e); |
|||
} |
|||
} |
|||
finally |
|||
{ |
|||
lock (_lock) |
|||
{ |
|||
_writeOperationPending = false; |
|||
} |
|||
} |
|||
} |
|||
|
|||
void FireException(Exception e) |
|||
{ |
|||
var cancel = e as OperationCanceledException; |
|||
if (cancel?.CancellationToken == _cancel) |
|||
return; |
|||
OnException?.Invoke(e); |
|||
} |
|||
|
|||
|
|||
public event Action<object> OnMessage; |
|||
public event Action<Exception> OnException; |
|||
} |
|||
} |
|||
@ -0,0 +1,27 @@ |
|||
using System; |
|||
using System.IO; |
|||
using System.Reflection; |
|||
|
|||
namespace Avalonia.Remote.Protocol |
|||
{ |
|||
public class BsonTcpTransport : TcpTransportBase |
|||
{ |
|||
public BsonTcpTransport(IMessageTypeResolver resolver) : base(resolver) |
|||
{ |
|||
} |
|||
|
|||
public BsonTcpTransport() : this(new DefaultMessageTypeResolver(typeof(BsonTcpTransport).GetTypeInfo().Assembly)) |
|||
{ |
|||
|
|||
} |
|||
|
|||
protected override IAvaloniaRemoteTransportConnection CreateTransport(IMessageTypeResolver resolver, |
|||
Stream stream, Action dispose) |
|||
{ |
|||
var t = new BsonStreamTransportConnection(resolver, stream, stream, dispose); |
|||
var wrap = new TransportConnectionWrapper(t); |
|||
t.StartReading(); |
|||
return wrap; |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,72 @@ |
|||
using System; |
|||
using System.Collections.Generic; |
|||
|
|||
namespace Avalonia.Remote.Protocol |
|||
{ |
|||
public class EventStash<T> |
|||
{ |
|||
private readonly Action<Exception> _exceptionHandler; |
|||
private List<T> _stash; |
|||
private Action<T> _delegate; |
|||
|
|||
public EventStash(Action<Exception> exceptionHandler = null) |
|||
{ |
|||
_exceptionHandler = exceptionHandler; |
|||
} |
|||
|
|||
public void Add(Action<T> handler) |
|||
{ |
|||
List<T> stash; |
|||
lock (this) |
|||
{ |
|||
var needsReplay = _delegate == null; |
|||
_delegate += handler; |
|||
if(!needsReplay) |
|||
return; |
|||
|
|||
lock (this) |
|||
{ |
|||
stash = _stash; |
|||
if(_stash == null) |
|||
return; |
|||
_stash = null; |
|||
} |
|||
} |
|||
foreach (var m in stash) |
|||
{ |
|||
if (_exceptionHandler != null) |
|||
try |
|||
{ |
|||
_delegate?.Invoke(m); |
|||
} |
|||
catch (Exception e) |
|||
{ |
|||
_exceptionHandler(e); |
|||
} |
|||
else |
|||
_delegate?.Invoke(m); |
|||
} |
|||
} |
|||
|
|||
|
|||
public void Remove(Action<T> handler) |
|||
{ |
|||
lock (this) |
|||
_delegate -= handler; |
|||
} |
|||
|
|||
public void Fire(T ev) |
|||
{ |
|||
if (_delegate == null) |
|||
{ |
|||
lock (this) |
|||
{ |
|||
_stash = _stash ?? new List<T>(); |
|||
_stash.Add(ev); |
|||
} |
|||
} |
|||
else |
|||
_delegate?.Invoke(ev); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,84 @@ |
|||
using System; |
|||
using System.IO; |
|||
using System.Net; |
|||
using System.Net.Sockets; |
|||
using System.Threading.Tasks; |
|||
|
|||
namespace Avalonia.Remote.Protocol |
|||
{ |
|||
public abstract class TcpTransportBase |
|||
{ |
|||
private readonly IMessageTypeResolver _resolver; |
|||
|
|||
public TcpTransportBase(IMessageTypeResolver resolver) |
|||
{ |
|||
_resolver = resolver; |
|||
} |
|||
|
|||
protected abstract IAvaloniaRemoteTransportConnection CreateTransport(IMessageTypeResolver resolver, |
|||
Stream stream, Action disposeCallback); |
|||
|
|||
class DisposableServer : IDisposable |
|||
{ |
|||
private readonly TcpListener _l; |
|||
|
|||
public DisposableServer(TcpListener l) |
|||
{ |
|||
_l = l; |
|||
} |
|||
public void Dispose() |
|||
{ |
|||
try |
|||
{ |
|||
_l.Stop(); |
|||
} |
|||
catch |
|||
{ |
|||
//Ignore
|
|||
} |
|||
} |
|||
} |
|||
|
|||
public IDisposable Listen(IPAddress address, int port, Action<IAvaloniaRemoteTransportConnection> cb) |
|||
{ |
|||
var server = new TcpListener(address, port); |
|||
async void AcceptNew() |
|||
{ |
|||
try |
|||
{ |
|||
var cl = await server.AcceptTcpClientAsync(); |
|||
AcceptNew(); |
|||
Task.Run(async () => |
|||
{ |
|||
try |
|||
{ |
|||
var tcs = new TaskCompletionSource<int>(); |
|||
var t = CreateTransport(_resolver, cl.GetStream(), () => tcs.TrySetResult(0)); |
|||
cb(t); |
|||
await tcs.Task; |
|||
} |
|||
finally |
|||
{ |
|||
cl.Dispose(); |
|||
} |
|||
|
|||
}); |
|||
} |
|||
catch |
|||
{ |
|||
//Ignore and stop
|
|||
} |
|||
} |
|||
server.Start(); |
|||
AcceptNew(); |
|||
return new DisposableServer(server); |
|||
} |
|||
|
|||
public async Task<IAvaloniaRemoteTransportConnection> Connect(IPAddress address, int port) |
|||
{ |
|||
var c = new TcpClient(); |
|||
await c.ConnectAsync(address, port); |
|||
return CreateTransport(_resolver, c.GetStream(), c.Dispose); |
|||
} |
|||
} |
|||
} |
|||
@ -1,7 +1,101 @@ |
|||
namespace Avalonia.Remote.Protocol |
|||
using System; |
|||
using System.Collections.Generic; |
|||
using System.Threading.Tasks; |
|||
|
|||
namespace Avalonia.Remote.Protocol |
|||
{ |
|||
public class TransportConnectionWrapper |
|||
public class TransportConnectionWrapper : IAvaloniaRemoteTransportConnection |
|||
{ |
|||
private readonly IAvaloniaRemoteTransportConnection _conn; |
|||
private EventStash<object> _onMessage; |
|||
private EventStash<Exception> _onException = new EventStash<Exception>(); |
|||
|
|||
private Queue<SendOperation> _sendQueue = new Queue<SendOperation>(); |
|||
private object _lock =new object(); |
|||
private TaskCompletionSource<int> _signal; |
|||
private bool _workerIsAlive; |
|||
public TransportConnectionWrapper(IAvaloniaRemoteTransportConnection conn) |
|||
{ |
|||
_conn = conn; |
|||
_onMessage = new EventStash<object>(_onException.Fire); |
|||
_conn.OnException +=_onException.Fire; |
|||
conn.OnMessage += _onMessage.Fire; |
|||
|
|||
} |
|||
|
|||
class SendOperation |
|||
{ |
|||
public object Message { get; set; } |
|||
public TaskCompletionSource<int> Tcs { get; set; } |
|||
} |
|||
|
|||
public void Dispose() => _conn.Dispose(); |
|||
|
|||
async void Worker() |
|||
{ |
|||
while (true) |
|||
{ |
|||
SendOperation wi = null; |
|||
lock (_lock) |
|||
{ |
|||
if (_sendQueue.Count != 0) |
|||
wi = _sendQueue.Dequeue(); |
|||
} |
|||
if (wi == null) |
|||
{ |
|||
var signal = new TaskCompletionSource<int>(); |
|||
lock (_lock) |
|||
_signal = signal; |
|||
await signal.Task.ConfigureAwait(false); |
|||
continue; |
|||
} |
|||
try |
|||
{ |
|||
await _conn.Send(wi.Message).ConfigureAwait(false); |
|||
wi.Tcs.TrySetResult(0); |
|||
} |
|||
catch (Exception e) |
|||
{ |
|||
wi.Tcs.TrySetException(e); |
|||
} |
|||
} |
|||
|
|||
} |
|||
|
|||
public Task Send(object data) |
|||
{ |
|||
var tcs = new TaskCompletionSource<int>(); |
|||
lock (_lock) |
|||
{ |
|||
if (!_workerIsAlive) |
|||
{ |
|||
_workerIsAlive = true; |
|||
Worker(); |
|||
} |
|||
_sendQueue.Enqueue(new SendOperation |
|||
{ |
|||
Message = data, |
|||
Tcs = tcs |
|||
}); |
|||
if (_signal != null) |
|||
{ |
|||
_signal.SetResult(0); |
|||
_signal = null; |
|||
} |
|||
} |
|||
return tcs.Task; |
|||
} |
|||
|
|||
public event Action<object> OnMessage |
|||
{ |
|||
add => _onMessage.Add(value); |
|||
remove => _onMessage.Remove(value); |
|||
} |
|||
|
|||
public event Action<Exception> OnException |
|||
{ |
|||
add => _onException.Add(value); |
|||
remove => _onException.Remove(value); |
|||
} |
|||
} |
|||
} |
|||
Loading…
Reference in new issue