Platform Networking

Realm/Realm/Native/AppConfiguration.cs

@@ -81,6 +81,8 @@ internal MetadataPersistenceMode? MetadataPersistence
 
         internal IntPtr managed_http_client;
 
+        internal IntPtr managed_websocket_provider;
+
         internal UInt64 sync_connect_timeout_ms;
 
         internal UInt64 sync_connection_linger_time_ms;

Realm/Realm/Native/PrimitiveValue.cs

@@ -283,13 +283,6 @@ public readonly bool TryGetObjectHandle(Realm realm, [NotNullWhen(true)] out Obj
             return false;
         }
 
-        [StructLayout(LayoutKind.Sequential)]
-        private struct BinaryValue
-        {
-            public byte* data;
-            public IntPtr size;
-        }
-
         [StructLayout(LayoutKind.Sequential)]
         private struct LinkValue
         {
@@ -348,4 +341,11 @@ public static StringValue AllocateFrom(string? value, Arena arena)
 
         public static implicit operator string?(in StringValue value) => !value ? null : Encoding.UTF8.GetString(value.data, (int)value.size);
     }
+
+    [StructLayout(LayoutKind.Sequential)]
+    internal unsafe struct BinaryValue
+    {
+        public byte* data;
+        public IntPtr size;
+    }
 }

Realm/Realm/Native/SyncSocketProvider.EventLoop.cs

@@ -0,0 +1,94 @@
+////////////////////////////////////////////////////////////////////////////
+//
+// Copyright 2023 Realm Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+////////////////////////////////////////////////////////////////////////////
+
+using System;
+using System.Text;
+using System.Threading;
+using System.Threading.Channels;
+using System.Threading.Tasks;
+
+namespace Realms.Native
+{
+    internal partial class SyncSocketProvider
+    {
+        private class Timer
+        {
+            private readonly CancellationTokenSource _cts = new();
+
+            internal Timer(TimeSpan delay, IntPtr nativeCallback, ChannelWriter<IWork> workQueue)
+            {
+                Task.Delay(delay).ContinueWith(t =>
+                {
+                    var status = Status.OK;
+                    if (t.IsCanceled)
+                    {
+                        status = new() { code = NativeMethods.ErrorCode.OperationAborted, reason = "Timer canceled" };
+                    }
+
+                    return workQueue.WriteAsync(new EventLoopWork(nativeCallback, status));
+                });
+            }
+
+            internal void Cancel()
+            {
+                _cts.Cancel();
+                _cts.Dispose();
+            }
+        }
+
+        private class EventLoopWork : IWork
+        {
+            private readonly IntPtr _nativeCallback;
+            private readonly Status _status;
+
+            public EventLoopWork(IntPtr nativeCallback, Status status)
+            {
+                _nativeCallback = nativeCallback;
+                _status = status;
+            }
+
+            public unsafe void Execute()
+            {
+                if (!string.IsNullOrEmpty(_status.reason))
+                {
+                    var bytes = Encoding.UTF8.GetBytes(_status.reason);
+                    fixed (byte* data = bytes)
+                    {
+                        var reason = new StringValue { data = data, size = bytes.Length };
+                        NativeMethods.run_callback(_nativeCallback, _status.code, reason);
+                    }
+                }
+                else
+                {
+                    NativeMethods.run_callback(_nativeCallback, _status.code, new());
+                }
+            }
+        }
+
+        private partial async Task WorkThread()
+        {
+            while (await _workQueue.Reader.WaitToReadAsync())
+            {
+                while (_workQueue.Reader.TryRead(out var work))
+                {
+                    work.Execute();
+                }
+            }
+        }
+    }
+}

Realm/Realm/Native/SyncSocketProvider.WebSocket.cs

@@ -0,0 +1,203 @@
+////////////////////////////////////////////////////////////////////////////
+//
+// Copyright 2023 Realm Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+////////////////////////////////////////////////////////////////////////////
+
+using System;
+using System.Buffers;
+using System.IO;
+using System.Net.WebSockets;
+using System.Runtime.InteropServices;
+using System.Text;
+using System.Threading.Channels;
+using System.Threading.Tasks;
+
+namespace Realms.Native
+{
+    internal partial class SyncSocketProvider
+    {
+        private class Socket : IDisposable
+        {
+            private readonly ClientWebSocket _webSocket;
+            private readonly IntPtr _observer;
+            private readonly ChannelWriter<IWork> _workQueue;
+
+            private readonly Uri _uri;
+            private readonly Task _workerThread;
+
+            private MemoryStream _receiveBuffer = new();
+
+            internal Socket(ClientWebSocket webSocket, ChannelWriter<IWork> workQueue, Uri uri) 
+            {
+                _webSocket = webSocket;
+                _workQueue = workQueue;
+                _uri = uri;
+                _workerThread = Task.Factory.StartNew(ReadThread, creationOptions: TaskCreationOptions.LongRunning | TaskCreationOptions.DenyChildAttach).Unwrap();
+            }
+
+            private async Task ReadThread()
+            {
+                try
+                {
+                    await _webSocket.ConnectAsync(_uri, default);
+                    await _workQueue.WriteAsync(new WebSocketConnectedWork(_observer, _webSocket.SubProtocol));
+                }
+                catch (WebSocketException e)
+                {
+                    // TODO: handle connect error
+                    return;
+                }
+
+                var buffer = new byte[32 * 1024];
+                do
+                {
+                    try
+                    {
+                        var result = await _webSocket.ReceiveAsync(new ArraySegment<byte>(buffer), default);
+                        if (result.MessageType == WebSocketMessageType.Binary)
+                        {
+                            await _receiveBuffer.WriteAsync(buffer, 0, result.Count);
+                            if (result.EndOfMessage)
+                            {
+                                var current_buffer = _receiveBuffer;
+                                _receiveBuffer = new MemoryStream();
+                                await _workQueue.WriteAsync(new BinaryMessageReceivedWork(current_buffer, _observer));
+                            }
+                        }
+                        else if (result.MessageType == WebSocketMessageType.Close)
+                        {
+                            await _workQueue.WriteAsync(new ClosedMessageReceivedWork(result.CloseStatus!.Value, result.CloseStatusDescription, _observer));
+                        }
+                    }
+                    catch(WebSocketException e)
+                    {
+                        // TODO: handle read error
+                    }
+                } while (_webSocket.State == WebSocketState.Open);
+            }
+
+            public void Write(BinaryValue data, IntPtr native_callback)
+            {
+                var buffer = ArrayPool<byte>.Shared.Rent((int)data.size);
+                unsafe
+                {
+                    Marshal.Copy((IntPtr)data.data, buffer, 0, (int)data.size);
+                }
+
+                _webSocket.SendAsync(new ArraySegment<byte>(buffer), WebSocketMessageType.Binary, true, default).ContinueWith(async t =>
+                {
+                    ArrayPool<byte>.Shared.Return(buffer);
+
+                    var status = Status.OK;
+                    if (t.IsFaulted)
+                    {
+                        // TODO: handle write error
+                    }
+
+                    await _workQueue.WriteAsync(new EventLoopWork(native_callback, status));
+                });
+            }
+
+            public void Dispose()
+            {
+                _webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, string.Empty, default).ContinueWith(t =>
+                {
+                    _webSocket.Dispose();
+                    _receiveBuffer.Dispose();
+                });
+            }
+        }
+
+        private abstract class WebSocketWork : IWork
+        {
+            private readonly IntPtr _observer;
+
+            protected WebSocketWork(IntPtr observer)
+            {
+                _observer = observer;
+            }
+
+            protected abstract void Execute(IntPtr observer);
+
+            void IWork.Execute()
+            {
+                Execute(_observer);
+            }
+        }
+
+        private sealed class WebSocketConnectedWork : WebSocketWork
+        {
+            private readonly string _protocol;
+
+            public WebSocketConnectedWork(IntPtr observer, string protocol)
+            : base(observer)
+            {
+                _protocol = protocol;
+            }
+
+            protected unsafe override void Execute(IntPtr observer)
+            {
+                var bytes = Encoding.UTF8.GetBytes(_protocol);
+                fixed (byte* data = bytes)
+                {
+                    NativeMethods.observer_connected_handler(observer, new() { data = data, size = bytes.Length });
+                }
+            }
+        }
+
+        private sealed class BinaryMessageReceivedWork : WebSocketWork
+        {
+            private MemoryStream _receiveBuffer;
+
+            public BinaryMessageReceivedWork(MemoryStream receiveBuffer, IntPtr observer)
+            : base(observer)
+            {
+                _receiveBuffer = receiveBuffer;
+            }
+
+            protected unsafe override void Execute(IntPtr observer)
+            {
+                using var buffer = _receiveBuffer;
+                fixed (byte* data = buffer.GetBuffer())
+                {
+                    NativeMethods.observer_binary_message_received(observer, new() { data = data, size = (IntPtr)buffer.Length });
+                }
+            }
+        }
+
+        private sealed class ClosedMessageReceivedWork : WebSocketWork
+        {
+            private WebSocketCloseStatus _status;
+            private string _description;
+
+            public ClosedMessageReceivedWork(WebSocketCloseStatus status, string description, IntPtr observer)
+            : base(observer)
+            {
+                _status = status;
+                _description = description;
+            }
+
+            protected unsafe override void Execute(IntPtr observer)
+            {
+                var bytes = Encoding.UTF8.GetBytes(_description);
+                fixed(byte* data = bytes)
+                {
+                    NativeMethods.observer_closed_handler(observer, _status, new() { data = data, size = bytes.Length });
+                }
+            }
+        }
+    }
+}