9875518284

Committed 10 Jul 2024 01:51PM UTC coverage: 81.384% (+0.03%) from 81.358%

Build # 9875518284

Build Type

Pull #3635

github

Specific Base a240be

Committed by

nirinchev

Commit Message

Hook up basic SG and weaver

Pull Request Pull Request #3635: Add flexible schema mapping POC

Run Details

2317 of 2997 branches covered (77.31%)

Branch coverage included in aggregate %.

12 of 15 new or added lines in 2 files covered. (80.0%)

7 existing lines in 1 file now uncovered.

6846 of 8262 relevant lines covered (82.86%)

42532.66 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

84.55

/Realm/Realm/Handles/SessionHandle.cs

﻿////////////////////////////////////////////////////////////////////////////
//
// Copyright 2016 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.Diagnostics;
using System.Linq;
using System.Runtime.InteropServices;
using System.Threading;
using System.Threading.Tasks;
using Realms.Exceptions;
using Realms.Exceptions.Sync;
using Realms.Logging;
using Realms.Native;
using Realms.Sync.ErrorHandling;
using Realms.Sync.Exceptions;
using Realms.Sync.Native;

namespace Realms.Sync
{
    internal class SessionHandle : RealmHandle
    {
        private static class NativeMethods
        {
            [UnmanagedFunctionPointer(CallingConvention.Cdecl)]
            public delegate void SessionErrorCallback(IntPtr session_handle_ptr,
                                                      SyncError error,
                                                      IntPtr managed_sync_config_handle);

            [UnmanagedFunctionPointer(CallingConvention.Cdecl)]
            public delegate void SessionProgressCallback(IntPtr progress_token_ptr, ulong transferred_bytes, ulong transferable_bytes, double progressEstimate);

            [UnmanagedFunctionPointer(CallingConvention.Cdecl)]
            public delegate void SessionWaitCallback(IntPtr task_completion_source, int error_code, PrimitiveValue message);

            [UnmanagedFunctionPointer(CallingConvention.Cdecl)]
            public delegate void SessionPropertyChangedCallback(IntPtr managed_session, NotifiableProperty property);

            [UnmanagedFunctionPointer(CallingConvention.Cdecl)]
            public delegate IntPtr NotifyBeforeClientReset(IntPtr before_frozen, IntPtr managed_sync_config_handle);

            [UnmanagedFunctionPointer(CallingConvention.Cdecl)]
            public delegate IntPtr NotifyAfterClientReset(IntPtr before_frozen, IntPtr after, IntPtr managed_sync_config_handle, bool did_recover);

            [DllImport(InteropConfig.DLL_NAME, EntryPoint = "realm_syncsession_install_callbacks", CallingConvention = CallingConvention.Cdecl)]
            public static extern void install_syncsession_callbacks(SessionErrorCallback error_callback,
                                                                    SessionProgressCallback progress_callback,
                                                                    SessionWaitCallback wait_callback,
                                                                    SessionPropertyChangedCallback property_changed_callback,
                                                                    NotifyBeforeClientReset notify_before,
                                                                    NotifyAfterClientReset notify_after);

            [DllImport(InteropConfig.DLL_NAME, EntryPoint = "realm_syncsession_get_user", CallingConvention = CallingConvention.Cdecl)]
            public static extern IntPtr get_user(SessionHandle session);

            [DllImport(InteropConfig.DLL_NAME, EntryPoint = "realm_syncsession_get_state", CallingConvention = CallingConvention.Cdecl)]
            public static extern SessionState get_state(SessionHandle session, out NativeException ex);

            [DllImport(InteropConfig.DLL_NAME, EntryPoint = "realm_syncsession_get_connection_state", CallingConvention = CallingConvention.Cdecl)]
            public static extern ConnectionState get_connection_state(SessionHandle session, out NativeException ex);

            [DllImport(InteropConfig.DLL_NAME, EntryPoint = "realm_syncsession_get_path", CallingConvention = CallingConvention.Cdecl)]
            public static extern IntPtr get_path(SessionHandle session, IntPtr buffer, IntPtr buffer_length, out NativeException ex);

            [DllImport(InteropConfig.DLL_NAME, EntryPoint = "realm_syncsession_get_raw_pointer", CallingConvention = CallingConvention.Cdecl)]
            public static extern IntPtr get_raw_pointer(SessionHandle session);

            [DllImport(InteropConfig.DLL_NAME, EntryPoint = "realm_syncsession_destroy", CallingConvention = CallingConvention.Cdecl)]
            public static extern void destroy(IntPtr handle);

            [DllImport(InteropConfig.DLL_NAME, EntryPoint = "realm_syncsession_register_progress_notifier", CallingConvention = CallingConvention.Cdecl)]
            public static extern ulong register_progress_notifier(SessionHandle session,
                                                                  IntPtr token_ptr,
                                                                  ProgressDirection direction,
                                                                  [MarshalAs(UnmanagedType.U1)] bool is_streaming,
                                                                  out NativeException ex);

            [DllImport(InteropConfig.DLL_NAME, EntryPoint = "realm_syncsession_unregister_progress_notifier", CallingConvention = CallingConvention.Cdecl)]
            public static extern void unregister_progress_notifier(SessionHandle session, ulong token, out NativeException ex);

            [DllImport(InteropConfig.DLL_NAME, EntryPoint = "realm_syncsession_register_property_changed_callback", CallingConvention = CallingConvention.Cdecl)]
            public static extern SessionNotificationToken register_property_changed_callback(SessionHandle session, IntPtr managed_session_handle, out NativeException ex);

            [DllImport(InteropConfig.DLL_NAME, EntryPoint = "realm_syncsession_unregister_property_changed_callback", CallingConvention = CallingConvention.Cdecl)]
            public static extern void unregister_property_changed_callback(IntPtr session, SessionNotificationToken token, out NativeException ex);

            [DllImport(InteropConfig.DLL_NAME, EntryPoint = "realm_syncsession_wait", CallingConvention = CallingConvention.Cdecl)]
            public static extern void wait(SessionHandle session, IntPtr task_completion_source, ProgressDirection direction, out NativeException ex);

            [DllImport(InteropConfig.DLL_NAME, EntryPoint = "realm_syncsession_report_error_for_testing", CallingConvention = CallingConvention.Cdecl)]
            public static extern void report_error_for_testing(SessionHandle session, int error_code, [MarshalAs(UnmanagedType.LPWStr)] string message, IntPtr message_len, [MarshalAs(UnmanagedType.U1)] bool is_fatal, int action);

            [DllImport(InteropConfig.DLL_NAME, EntryPoint = "realm_syncsession_stop", CallingConvention = CallingConvention.Cdecl)]
            public static extern void stop(SessionHandle session, out NativeException ex);

            [DllImport(InteropConfig.DLL_NAME, EntryPoint = "realm_syncsession_shutdown_and_wait", CallingConvention = CallingConvention.Cdecl)]
            public static extern void shutdown_and_wait(SessionHandle session, out NativeException ex);

            [DllImport(InteropConfig.DLL_NAME, EntryPoint = "realm_syncsession_start", CallingConvention = CallingConvention.Cdecl)]
            public static extern void start(SessionHandle session, out NativeException ex);
        }

        private SessionNotificationToken? _notificationToken;

        public override bool ForceRootOwnership => true;

        [Preserve]
        public SessionHandle(SharedRealmHandle? root, IntPtr handle) : base(root, handle)
        {
        }

        public static void Initialize()
        {
            NativeMethods.SessionErrorCallback error = HandleSessionError;
            NativeMethods.SessionProgressCallback progress = HandleSessionProgress;
            NativeMethods.SessionWaitCallback wait = HandleSessionWaitCallback;
            NativeMethods.SessionPropertyChangedCallback propertyChanged = HandleSessionPropertyChangedCallback;
            NativeMethods.NotifyBeforeClientReset beforeReset = NotifyBeforeClientReset;
            NativeMethods.NotifyAfterClientReset afterReset = NotifyAfterClientReset;

            GCHandle.Alloc(error);
            GCHandle.Alloc(progress);
            GCHandle.Alloc(wait);
            GCHandle.Alloc(propertyChanged);
            GCHandle.Alloc(beforeReset);
            GCHandle.Alloc(afterReset);

            NativeMethods.install_syncsession_callbacks(error, progress, wait, propertyChanged, beforeReset, afterReset);
        }

        public SyncUserHandle GetUser()
        {
            var ptr = NativeMethods.get_user(this);
            if (ptr == IntPtr.Zero)
            {
                throw new RealmException("Unable to obtain user for session. This likely means the session is being torn down.");
            }

            return new(ptr);
        }

        public SessionState GetState()
        {
            var state = NativeMethods.get_state(this, out var ex);
            ex.ThrowIfNecessary();
            return state;
        }

        public ConnectionState GetConnectionState()
        {
            var connectionState = NativeMethods.get_connection_state(this, out var ex);
            ex.ThrowIfNecessary();
            return connectionState;
        }

        public string GetPath()
        {
            return MarshalHelpers.GetString((IntPtr buffer, IntPtr length, out bool isNull, out NativeException ex) =>
            {
                isNull = false;
                return NativeMethods.get_path(this, buffer, length, out ex);
            })!;
        }

        public ulong RegisterProgressNotifier(GCHandle managedHandle, ProgressDirection direction, ProgressMode mode)
        {
            var isStreaming = mode == ProgressMode.ReportIndefinitely;
            var token = NativeMethods.register_progress_notifier(this, GCHandle.ToIntPtr(managedHandle), direction, isStreaming, out var ex);
            ex.ThrowIfNecessary();
            return token;
        }

        public void UnregisterProgressNotifier(ulong token)
        {
            NativeMethods.unregister_progress_notifier(this, token, out var ex);
            ex.ThrowIfNecessary();
        }

        public void SubscribeNotifications(Session session)
        {
            Debug.Assert(!_notificationToken.HasValue, $"{nameof(_notificationToken)} must be null before subscribing.");

            var managedSessionHandle = GCHandle.Alloc(session, GCHandleType.Weak);
            var sessionPointer = GCHandle.ToIntPtr(managedSessionHandle);
            _notificationToken = NativeMethods.register_property_changed_callback(this, sessionPointer, out var ex);
            ex.ThrowIfNecessary();
        }

        public void UnsubscribeNotifications()
        {
            if (_notificationToken.HasValue)
            {
                // This needs to use the handle directly because it's being called in Unbind. At this point the SafeHandle is closed, which means we'll
                // get an error if we attempted to marshal it to native. The raw pointer is fine though and we can use it.
                NativeMethods.unregister_property_changed_callback(handle, _notificationToken.Value, out var ex);
                _notificationToken = null;
                ex.ThrowIfNecessary();
            }
        }

        public Task WaitAsync(ProgressDirection direction, CancellationToken? cancellationToken)
        {
            var tcs = new TaskCompletionSource();
            if (cancellationToken?.IsCancellationRequested == true)
            {
                tcs.TrySetCanceled(cancellationToken.Value);
                return tcs.Task;
            }

            // The tcsHandles is freed in HandleSessionWaitCallback. It's important that we don't free it on cancellation
            // as the cancellation doesn't really cancel the native wait operation. That will eventually complete and it needs
            // to have the GCHandle at this point, otherwise we'll get a hard crash on Mono.
            var tcsHandle = GCHandle.Alloc(tcs);

            cancellationToken?.Register(() => tcs.TrySetCanceled(cancellationToken.Value));

            try
            {
                NativeMethods.wait(this, GCHandle.ToIntPtr(tcsHandle), direction, out var ex);
                ex.ThrowIfNecessary();
            }
            catch
            {
                // If we failed to register a waiter, we can free the handle as we won't get a native callback here anyway
                tcsHandle.Free();
                throw;
            }

            return tcs.Task;
        }

        public IntPtr GetRawPointer()
        {
            return NativeMethods.get_raw_pointer(this);
        }

        public void ReportErrorForTesting(int errorCode, string errorMessage, bool isFatal, ServerRequestsAction action)
        {
            NativeMethods.report_error_for_testing(this, errorCode, errorMessage, (IntPtr)errorMessage.Length, isFatal, (int)action);
        }

        public void Stop()
        {
            NativeMethods.stop(this, out var ex);
            ex.ThrowIfNecessary();
        }

        public void Start()
        {
            NativeMethods.start(this, out var ex);
            ex.ThrowIfNecessary();
        }

        /// <summary>
        /// Terminates the sync session and releases the Realm file it was using.
        /// </summary>
        public void ShutdownAndWait()
        {
            NativeMethods.shutdown_and_wait(this, out var ex);
            ex.ThrowIfNecessary();
        }

        public override void Unbind()
        {
            UnsubscribeNotifications();
            NativeMethods.destroy(handle);
        }

        [MonoPInvokeCallback(typeof(NativeMethods.SessionErrorCallback))]
        private static void HandleSessionError(IntPtr sessionHandlePtr, SyncError error, IntPtr managedSyncConfigurationBaseHandle)
        {
            try
            {
                // Filter out end of input, which the client seems to have started reporting
                if (error.error_code == (ErrorCode)1)
                {
                    return;
                }

                using var handle = new SessionHandle(null, sessionHandlePtr);
                var session = new Session(handle);
                string messageString = error.message!;
                var syncConfigHandle = GCHandle.FromIntPtr(managedSyncConfigurationBaseHandle);
                var syncConfig = (SyncConfigurationBase)syncConfigHandle.Target!;

                if (error.is_client_reset)
                {
                    var userInfo = error.user_info_pairs.ToEnumerable().ToDictionary(kvp => (string)kvp.Key!, kvp => (string?)kvp.Value);
                    var clientResetEx = new ClientResetException(session.User.App, messageString, error.error_code, userInfo);

                    syncConfig.ClientResetHandler.ManualClientReset?.Invoke(clientResetEx);
                    return;
                }

                SessionException exception;
                if (error.error_code == ErrorCode.CompensatingWrite)
                {
                    var compensatingWrites = error.compensating_writes
                        .ToEnumerable()
                        .Select(c => new CompensatingWriteInfo(c.object_name!, c.reason!, new RealmValue(c.primary_key)))
                        .ToArray();
                    exception = new CompensatingWriteException(messageString, compensatingWrites);
                }
                else
                {
                    exception = new SessionException(messageString, error.error_code);
                }

                exception.HelpLink = error.log_url;
                syncConfig.OnSessionError?.Invoke(session, exception);
            }
            catch (Exception ex)
            {
                Logger.Default.Log(LogLevel.Warn, $"An error has occurred while handling a session error: {ex}");
            }
        }

        [MonoPInvokeCallback(typeof(NativeMethods.NotifyBeforeClientReset))]
        private static IntPtr NotifyBeforeClientReset(IntPtr beforeFrozen, IntPtr managedSyncConfigurationHandle)
        {
            SyncConfigurationBase? syncConfig = null;

            try
            {
                var syncConfigHandle = GCHandle.FromIntPtr(managedSyncConfigurationHandle);
                syncConfig = (SyncConfigurationBase)syncConfigHandle.Target!;

                var cb = syncConfig.ClientResetHandler switch
                {
                    DiscardUnsyncedChangesHandler handler => handler.OnBeforeReset,
                    RecoverUnsyncedChangesHandler handler => handler.OnBeforeReset,
                    RecoverOrDiscardUnsyncedChangesHandler handler => handler.OnBeforeReset,
                    _ => throw new NotSupportedException($"ClientResetHandlerBase of type {syncConfig.ClientResetHandler.GetType()} is not handled yet")
                };

                if (cb != null)
                {
                    var schema = syncConfig.Schema;
                    using var realmBefore = new Realm(new UnownedRealmHandle(beforeFrozen), syncConfig, schema);
                    cb.Invoke(realmBefore);
                }

                return IntPtr.Zero;
            }
            catch (Exception ex)
            {
                var handlerType = syncConfig is null ? "ClientResetHandler" : syncConfig.ClientResetHandler.GetType().Name;
                Logger.Default.Log(LogLevel.Error, $"An error has occurred while executing {handlerType}.OnBeforeReset during a client reset: {ex}");

                var exHandle = GCHandle.Alloc(ex);
                return GCHandle.ToIntPtr(exHandle);
            }
        }

        [MonoPInvokeCallback(typeof(NativeMethods.NotifyAfterClientReset))]
        private static IntPtr NotifyAfterClientReset(IntPtr beforeFrozen, IntPtr after, IntPtr managedSyncConfigurationHandle, bool didRecover)
        {
            SyncConfigurationBase? syncConfig = null;

            try
            {
                var syncConfigHandle = GCHandle.FromIntPtr(managedSyncConfigurationHandle);
                syncConfig = (SyncConfigurationBase)syncConfigHandle.Target!;

                var cb = syncConfig.ClientResetHandler switch
                {
                    DiscardUnsyncedChangesHandler handler => handler.OnAfterReset,
                    RecoverUnsyncedChangesHandler handler => handler.OnAfterReset,
                    RecoverOrDiscardUnsyncedChangesHandler handler => didRecover ? handler.OnAfterRecovery : handler.OnAfterDiscard,
                    _ => throw new NotSupportedException($"ClientResetHandlerBase of type {syncConfig.ClientResetHandler.GetType()} is not handled yet")
                };

                if (cb != null)
                {
                    var schema = syncConfig.Schema;
                    using var realmBefore = new Realm(new UnownedRealmHandle(beforeFrozen), syncConfig, schema);
                    using var realmAfter = new Realm(new UnownedRealmHandle(after), syncConfig, schema);
                    cb.Invoke(realmBefore, realmAfter);
                }

                return IntPtr.Zero;
            }
            catch (Exception ex)
            {
                var handlerType = syncConfig is null ? "ClientResetHandler" : syncConfig.ClientResetHandler.GetType().Name;
                Logger.Default.Log(LogLevel.Error, $"An error has occurred while executing {handlerType}.OnAfterReset during a client reset: {ex}");

                var exHandle = GCHandle.Alloc(ex);
                return GCHandle.ToIntPtr(exHandle);
            }
        }

        [MonoPInvokeCallback(typeof(NativeMethods.SessionProgressCallback))]
        private static void HandleSessionProgress(IntPtr tokenPtr, ulong transferredBytes, ulong transferableBytes, double progressEstimate)
        {
            var token = (ProgressNotificationToken?)GCHandle.FromIntPtr(tokenPtr).Target;

            // This is used to provide a reasonable progress estimate until the core work is done
            double managedProgressEstimate = transferableBytes > 0.0 ? transferredBytes / transferableBytes : 1.0;
            token?.Notify(managedProgressEstimate);
        }

        [MonoPInvokeCallback(typeof(NativeMethods.SessionWaitCallback))]
        private static void HandleSessionWaitCallback(IntPtr taskCompletionSource, int error_code, PrimitiveValue message)
        {
            var handle = GCHandle.FromIntPtr(taskCompletionSource);
            var tcs = (TaskCompletionSource)handle.Target!;

            if (error_code == 0)
            {
                tcs.TrySetResult();
            }
            else
            {
                var inner = new SessionException(message.AsString(), (ErrorCode)error_code);
                const string OuterMessage = "A system error occurred while waiting for completion. See InnerException for more details";
                tcs.TrySetException(new RealmException(OuterMessage, inner));
            }

            handle.Free();
        }

        [MonoPInvokeCallback(typeof(NativeMethods.SessionPropertyChangedCallback))]
        private static void HandleSessionPropertyChangedCallback(IntPtr managedSessionHandle, NotifiableProperty property)
        {
            try
            {
                if (managedSessionHandle == IntPtr.Zero)
                {
                    return;
                }

                var propertyName = property switch
                {
                    NotifiableProperty.ConnectionState => nameof(Session.ConnectionState),
                    _ => throw new NotSupportedException($"Unexpected notifiable property value: {property}")
                };
                var session = (Session?)GCHandle.FromIntPtr(managedSessionHandle).Target;
                if (session is null)
                {
                    // We're taking a weak handle to the session, so it's possible that it's been collected
                    return;
                }

                ThreadPool.QueueUserWorkItem(_ =>
                {
                    session.RaisePropertyChanged(propertyName);
                });
            }
            catch (Exception ex)
            {
                Logger.Default.Log(LogLevel.Error, $"An error has occurred while raising a property changed event: {ex}");
            }
        }

        private enum NotifiableProperty : byte
        {
            ConnectionState = 0
        }
    }
}

1	////////////////////////////////////////////////////////////////////////////
2	//
3	// Copyright 2016 Realm Inc.
4	//
5	// Licensed under the Apache License, Version 2.0 (the "License");
6	// you may not use this file except in compliance with the License.
7	// You may obtain a copy of the License at
8	//
9	// http://www.apache.org/licenses/LICENSE-2.0
10	//
11	// Unless required by applicable law or agreed to in writing, software
12	// distributed under the License is distributed on an "AS IS" BASIS,
13	// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14	// See the License for the specific language governing permissions and
15	// limitations under the License.
16	//
17	////////////////////////////////////////////////////////////////////////////
18
19	using System;
20	using System.Diagnostics;
21	using System.Linq;
22	using System.Runtime.InteropServices;
23	using System.Threading;
24	using System.Threading.Tasks;
25	using Realms.Exceptions;
26	using Realms.Exceptions.Sync;
27	using Realms.Logging;
28	using Realms.Native;
29	using Realms.Sync.ErrorHandling;
30	using Realms.Sync.Exceptions;
31	using Realms.Sync.Native;
32
33	namespace Realms.Sync
34	{
35	internal class SessionHandle : RealmHandle
36	{
37	private static class NativeMethods
38	{
39	[UnmanagedFunctionPointer(CallingConvention.Cdecl)]
40	public delegate void SessionErrorCallback(IntPtr session_handle_ptr,
41	SyncError error,
42	IntPtr managed_sync_config_handle);
43
44	[UnmanagedFunctionPointer(CallingConvention.Cdecl)]
45	public delegate void SessionProgressCallback(IntPtr progress_token_ptr, ulong transferred_bytes, ulong transferable_bytes, double progressEstimate);
46
47	[UnmanagedFunctionPointer(CallingConvention.Cdecl)]
48	public delegate void SessionWaitCallback(IntPtr task_completion_source, int error_code, PrimitiveValue message);
49
50	[UnmanagedFunctionPointer(CallingConvention.Cdecl)]
51	public delegate void SessionPropertyChangedCallback(IntPtr managed_session, NotifiableProperty property);
52
53	[UnmanagedFunctionPointer(CallingConvention.Cdecl)]
54	public delegate IntPtr NotifyBeforeClientReset(IntPtr before_frozen, IntPtr managed_sync_config_handle);
55
56	[UnmanagedFunctionPointer(CallingConvention.Cdecl)]
57	public delegate IntPtr NotifyAfterClientReset(IntPtr before_frozen, IntPtr after, IntPtr managed_sync_config_handle, bool did_recover);
58
59	[DllImport(InteropConfig.DLL_NAME, EntryPoint = "realm_syncsession_install_callbacks", CallingConvention = CallingConvention.Cdecl)]
60	public static extern void install_syncsession_callbacks(SessionErrorCallback error_callback,
61	SessionProgressCallback progress_callback,
62	SessionWaitCallback wait_callback,
63	SessionPropertyChangedCallback property_changed_callback,
64	NotifyBeforeClientReset notify_before,
65	NotifyAfterClientReset notify_after);
66
67	[DllImport(InteropConfig.DLL_NAME, EntryPoint = "realm_syncsession_get_user", CallingConvention = CallingConvention.Cdecl)]
68	public static extern IntPtr get_user(SessionHandle session);
69
70	[DllImport(InteropConfig.DLL_NAME, EntryPoint = "realm_syncsession_get_state", CallingConvention = CallingConvention.Cdecl)]
71	public static extern SessionState get_state(SessionHandle session, out NativeException ex);
72
73	[DllImport(InteropConfig.DLL_NAME, EntryPoint = "realm_syncsession_get_connection_state", CallingConvention = CallingConvention.Cdecl)]
74	public static extern ConnectionState get_connection_state(SessionHandle session, out NativeException ex);
75
76	[DllImport(InteropConfig.DLL_NAME, EntryPoint = "realm_syncsession_get_path", CallingConvention = CallingConvention.Cdecl)]
77	public static extern IntPtr get_path(SessionHandle session, IntPtr buffer, IntPtr buffer_length, out NativeException ex);
78
79	[DllImport(InteropConfig.DLL_NAME, EntryPoint = "realm_syncsession_get_raw_pointer", CallingConvention = CallingConvention.Cdecl)]
80	public static extern IntPtr get_raw_pointer(SessionHandle session);
81
82	[DllImport(InteropConfig.DLL_NAME, EntryPoint = "realm_syncsession_destroy", CallingConvention = CallingConvention.Cdecl)]
83	public static extern void destroy(IntPtr handle);
84
85	[DllImport(InteropConfig.DLL_NAME, EntryPoint = "realm_syncsession_register_progress_notifier", CallingConvention = CallingConvention.Cdecl)]
86	public static extern ulong register_progress_notifier(SessionHandle session,
87	IntPtr token_ptr,
88	ProgressDirection direction,
89	[MarshalAs(UnmanagedType.U1)] bool is_streaming,
90	out NativeException ex);
91
92	[DllImport(InteropConfig.DLL_NAME, EntryPoint = "realm_syncsession_unregister_progress_notifier", CallingConvention = CallingConvention.Cdecl)]
93	public static extern void unregister_progress_notifier(SessionHandle session, ulong token, out NativeException ex);
94
95	[DllImport(InteropConfig.DLL_NAME, EntryPoint = "realm_syncsession_register_property_changed_callback", CallingConvention = CallingConvention.Cdecl)]
96	public static extern SessionNotificationToken register_property_changed_callback(SessionHandle session, IntPtr managed_session_handle, out NativeException ex);
97
98	[DllImport(InteropConfig.DLL_NAME, EntryPoint = "realm_syncsession_unregister_property_changed_callback", CallingConvention = CallingConvention.Cdecl)]
99	public static extern void unregister_property_changed_callback(IntPtr session, SessionNotificationToken token, out NativeException ex);
100
101	[DllImport(InteropConfig.DLL_NAME, EntryPoint = "realm_syncsession_wait", CallingConvention = CallingConvention.Cdecl)]
102	public static extern void wait(SessionHandle session, IntPtr task_completion_source, ProgressDirection direction, out NativeException ex);
103
104	[DllImport(InteropConfig.DLL_NAME, EntryPoint = "realm_syncsession_report_error_for_testing", CallingConvention = CallingConvention.Cdecl)]
105	public static extern void report_error_for_testing(SessionHandle session, int error_code, [MarshalAs(UnmanagedType.LPWStr)] string message, IntPtr message_len, [MarshalAs(UnmanagedType.U1)] bool is_fatal, int action);
106
107	[DllImport(InteropConfig.DLL_NAME, EntryPoint = "realm_syncsession_stop", CallingConvention = CallingConvention.Cdecl)]
108	public static extern void stop(SessionHandle session, out NativeException ex);
109
110	[DllImport(InteropConfig.DLL_NAME, EntryPoint = "realm_syncsession_shutdown_and_wait", CallingConvention = CallingConvention.Cdecl)]
111	public static extern void shutdown_and_wait(SessionHandle session, out NativeException ex);
112
113	[DllImport(InteropConfig.DLL_NAME, EntryPoint = "realm_syncsession_start", CallingConvention = CallingConvention.Cdecl)]
114	public static extern void start(SessionHandle session, out NativeException ex);
115	}
116
117	private SessionNotificationToken? _notificationToken;
118
119	public override bool ForceRootOwnership => true;	486✔
120
121	[Preserve]
122	public SessionHandle(SharedRealmHandle? root, IntPtr handle) : base(root, handle)	512✔
123	{
124	}	512✔
125
126	public static void Initialize()
127	{
128	NativeMethods.SessionErrorCallback error = HandleSessionError;	1✔
129	NativeMethods.SessionProgressCallback progress = HandleSessionProgress;	1✔
130	NativeMethods.SessionWaitCallback wait = HandleSessionWaitCallback;	1✔
131	NativeMethods.SessionPropertyChangedCallback propertyChanged = HandleSessionPropertyChangedCallback;	1✔
132	NativeMethods.NotifyBeforeClientReset beforeReset = NotifyBeforeClientReset;	1✔
133	NativeMethods.NotifyAfterClientReset afterReset = NotifyAfterClientReset;	1✔
134
135	GCHandle.Alloc(error);	1✔
136	GCHandle.Alloc(progress);	1✔
137	GCHandle.Alloc(wait);	1✔
138	GCHandle.Alloc(propertyChanged);	1✔
139	GCHandle.Alloc(beforeReset);	1✔
140	GCHandle.Alloc(afterReset);	1✔
141
142	NativeMethods.install_syncsession_callbacks(error, progress, wait, propertyChanged, beforeReset, afterReset);	1✔
143	}	1✔
144
145	public SyncUserHandle GetUser()
146	{
147	var ptr = NativeMethods.get_user(this);	23✔
148	if (ptr == IntPtr.Zero)	23!
149	{
150	throw new RealmException("Unable to obtain user for session. This likely means the session is being torn down.");	×
151	}
152
153	return new(ptr);	23✔
154	}
155
156	public SessionState GetState()
157	{
158	var state = NativeMethods.get_state(this, out var ex);	16✔
159	ex.ThrowIfNecessary();	16✔
160	return state;	16✔
161	}
162
163	public ConnectionState GetConnectionState()
164	{
165	var connectionState = NativeMethods.get_connection_state(this, out var ex);	12✔
166	ex.ThrowIfNecessary();	12✔
167	return connectionState;	12✔
168	}
169
170	public string GetPath()
171	{
172	return MarshalHelpers.GetString((IntPtr buffer, IntPtr length, out bool isNull, out NativeException ex) =>	×
173	{	×
174	isNull = false;	×
175	return NativeMethods.get_path(this, buffer, length, out ex);	×
176	})!;	×
177	}
178
179	public ulong RegisterProgressNotifier(GCHandle managedHandle, ProgressDirection direction, ProgressMode mode)
180	{
181	var isStreaming = mode == ProgressMode.ReportIndefinitely;	4✔
182	var token = NativeMethods.register_progress_notifier(this, GCHandle.ToIntPtr(managedHandle), direction, isStreaming, out var ex);	4✔
183	ex.ThrowIfNecessary();	4✔
184	return token;	4✔
185	}
186
187	public void UnregisterProgressNotifier(ulong token)
188	{
189	NativeMethods.unregister_progress_notifier(this, token, out var ex);	4✔
190	ex.ThrowIfNecessary();	4✔
191	}	4✔
192
193	public void SubscribeNotifications(Session session)
194	{
195	Debug.Assert(!_notificationToken.HasValue, $"{nameof(_notificationToken)} must be null before subscribing.");
196
197	var managedSessionHandle = GCHandle.Alloc(session, GCHandleType.Weak);	9✔
198	var sessionPointer = GCHandle.ToIntPtr(managedSessionHandle);	9✔
199	_notificationToken = NativeMethods.register_property_changed_callback(this, sessionPointer, out var ex);	9✔
200	ex.ThrowIfNecessary();	9✔
201	}	9✔
202
203	public void UnsubscribeNotifications()
204	{
205	if (_notificationToken.HasValue)	519✔
206	{
207	// This needs to use the handle directly because it's being called in Unbind. At this point the SafeHandle is closed, which means we'll
208	// get an error if we attempted to marshal it to native. The raw pointer is fine though and we can use it.
209	NativeMethods.unregister_property_changed_callback(handle, _notificationToken.Value, out var ex);	9✔
210	_notificationToken = null;	9✔
211	ex.ThrowIfNecessary();	9✔
212	}
213	}	519✔
214
215	public Task WaitAsync(ProgressDirection direction, CancellationToken? cancellationToken)
216	{
217	var tcs = new TaskCompletionSource();	567✔
218	if (cancellationToken?.IsCancellationRequested == true)	567✔
219	{
220	tcs.TrySetCanceled(cancellationToken.Value);	2✔
221	return tcs.Task;	2✔
222	}
223
224	// The tcsHandles is freed in HandleSessionWaitCallback. It's important that we don't free it on cancellation
225	// as the cancellation doesn't really cancel the native wait operation. That will eventually complete and it needs
226	// to have the GCHandle at this point, otherwise we'll get a hard crash on Mono.
227	var tcsHandle = GCHandle.Alloc(tcs);	565✔
228
229	cancellationToken?.Register(() => tcs.TrySetCanceled(cancellationToken.Value));	567✔
230
231	try
232	{
233	NativeMethods.wait(this, GCHandle.ToIntPtr(tcsHandle), direction, out var ex);	565✔
234	ex.ThrowIfNecessary();	565✔
235	}	565✔
236	catch	×
237	{
238	// If we failed to register a waiter, we can free the handle as we won't get a native callback here anyway
239	tcsHandle.Free();	×
240	throw;	×
241	}
242
243	return tcs.Task;	565✔
244	}
245
246	public IntPtr GetRawPointer()
247	{
248	return NativeMethods.get_raw_pointer(this);	14✔
249	}
250
251	public void ReportErrorForTesting(int errorCode, string errorMessage, bool isFatal, ServerRequestsAction action)
252	{
253	NativeMethods.report_error_for_testing(this, errorCode, errorMessage, (IntPtr)errorMessage.Length, isFatal, (int)action);	1✔
254	}	1✔
255
256	public void Stop()
257	{
258	NativeMethods.stop(this, out var ex);	72✔
259	ex.ThrowIfNecessary();	72✔
260	}	72✔
261
262	public void Start()
263	{
264	NativeMethods.start(this, out var ex);	46✔
265	ex.ThrowIfNecessary();	46✔
266	}	46✔
267
268	/// <summary>
269	/// Terminates the sync session and releases the Realm file it was using.
270	/// </summary>
271	public void ShutdownAndWait()
272	{
273	NativeMethods.shutdown_and_wait(this, out var ex);	4✔
274	ex.ThrowIfNecessary();	4✔
275	}	4✔
276
277	public override void Unbind()
278	{
279	UnsubscribeNotifications();	512✔
280	NativeMethods.destroy(handle);	512✔
281	}	512✔
282
283	[MonoPInvokeCallback(typeof(NativeMethods.SessionErrorCallback))]
284	private static void HandleSessionError(IntPtr sessionHandlePtr, SyncError error, IntPtr managedSyncConfigurationBaseHandle)
285	{
286	try
287	{
288	// Filter out end of input, which the client seems to have started reporting
289	if (error.error_code == (ErrorCode)1)	26!
290	{
291	return;	×
292	}
293
294	using var handle = new SessionHandle(null, sessionHandlePtr);	26✔
295	var session = new Session(handle);	26✔
296	string messageString = error.message!;	26✔
297	var syncConfigHandle = GCHandle.FromIntPtr(managedSyncConfigurationBaseHandle);	26✔
298	var syncConfig = (SyncConfigurationBase)syncConfigHandle.Target!;	26✔
299
300	if (error.is_client_reset)	26✔
301	{
302	var userInfo = error.user_info_pairs.ToEnumerable().ToDictionary(kvp => (string)kvp.Key!, kvp => (string?)kvp.Value);	100✔
303	var clientResetEx = new ClientResetException(session.User.App, messageString, error.error_code, userInfo);	20✔
304
305	syncConfig.ClientResetHandler.ManualClientReset?.Invoke(clientResetEx);	20!
306	return;	20✔
307	}
308
309	SessionException exception;
310	if (error.error_code == ErrorCode.CompensatingWrite)	6✔
311	{
312	var compensatingWrites = error.compensating_writes	3✔
313	.ToEnumerable()	3✔
314	.Select(c => new CompensatingWriteInfo(c.object_name!, c.reason!, new RealmValue(c.primary_key)))	3✔
315	.ToArray();	3✔
316	exception = new CompensatingWriteException(messageString, compensatingWrites);	3✔
317	}
318	else
319	{
320	exception = new SessionException(messageString, error.error_code);	3✔
321	}
322
323	exception.HelpLink = error.log_url;	6✔
324	syncConfig.OnSessionError?.Invoke(session, exception);	6✔
325	}	6✔
326	catch (Exception ex)	×
327	{
328	Logger.Default.Log(LogLevel.Warn, $"An error has occurred while handling a session error: {ex}");	×
329	}	×
330	}	26✔
331
332	[MonoPInvokeCallback(typeof(NativeMethods.NotifyBeforeClientReset))]
333	private static IntPtr NotifyBeforeClientReset(IntPtr beforeFrozen, IntPtr managedSyncConfigurationHandle)
334	{
335	SyncConfigurationBase? syncConfig = null;	43✔
336
337	try
338	{
339	var syncConfigHandle = GCHandle.FromIntPtr(managedSyncConfigurationHandle);	43✔
340	syncConfig = (SyncConfigurationBase)syncConfigHandle.Target!;	43✔
341
342	var cb = syncConfig.ClientResetHandler switch	43!
343	{	43✔
344	DiscardUnsyncedChangesHandler handler => handler.OnBeforeReset,	13✔
345	RecoverUnsyncedChangesHandler handler => handler.OnBeforeReset,	12✔
346	RecoverOrDiscardUnsyncedChangesHandler handler => handler.OnBeforeReset,	18✔
347	_ => throw new NotSupportedException($"ClientResetHandlerBase of type {syncConfig.ClientResetHandler.GetType()} is not handled yet")	×
348	};	43✔
349
350	if (cb != null)	43✔
351	{
352	var schema = syncConfig.Schema;	30✔
353	using var realmBefore = new Realm(new UnownedRealmHandle(beforeFrozen), syncConfig, schema);	30✔
354	cb.Invoke(realmBefore);	30✔
355	}
356
357	return IntPtr.Zero;	31✔
358	}
359	catch (Exception ex)	12✔
360	{
361	var handlerType = syncConfig is null ? "ClientResetHandler" : syncConfig.ClientResetHandler.GetType().Name;	12!
362	Logger.Default.Log(LogLevel.Error, $"An error has occurred while executing {handlerType}.OnBeforeReset during a client reset: {ex}");	12✔
363
364	var exHandle = GCHandle.Alloc(ex);	12✔
365	return GCHandle.ToIntPtr(exHandle);	12✔
366	}
367	}	43✔
368
369	[MonoPInvokeCallback(typeof(NativeMethods.NotifyAfterClientReset))]
370	private static IntPtr NotifyAfterClientReset(IntPtr beforeFrozen, IntPtr after, IntPtr managedSyncConfigurationHandle, bool didRecover)
371	{
372	SyncConfigurationBase? syncConfig = null;	31✔
373
374	try
375	{
376	var syncConfigHandle = GCHandle.FromIntPtr(managedSyncConfigurationHandle);	31✔
377	syncConfig = (SyncConfigurationBase)syncConfigHandle.Target!;	31✔
378
379	var cb = syncConfig.ClientResetHandler switch	31!
380	{	31✔
381	DiscardUnsyncedChangesHandler handler => handler.OnAfterReset,	9✔
382	RecoverUnsyncedChangesHandler handler => handler.OnAfterReset,	8✔
383	RecoverOrDiscardUnsyncedChangesHandler handler => didRecover ? handler.OnAfterRecovery : handler.OnAfterDiscard,	14✔
384	_ => throw new NotSupportedException($"ClientResetHandlerBase of type {syncConfig.ClientResetHandler.GetType()} is not handled yet")	×
385	};	31✔
386
387	if (cb != null)	31✔
388	{
389	var schema = syncConfig.Schema;	20✔
390	using var realmBefore = new Realm(new UnownedRealmHandle(beforeFrozen), syncConfig, schema);	20✔
391	using var realmAfter = new Realm(new UnownedRealmHandle(after), syncConfig, schema);	20✔
392	cb.Invoke(realmBefore, realmAfter);	20✔
393	}
394
395	return IntPtr.Zero;	25✔
396	}
397	catch (Exception ex)	6✔
398	{
399	var handlerType = syncConfig is null ? "ClientResetHandler" : syncConfig.ClientResetHandler.GetType().Name;	6!
400	Logger.Default.Log(LogLevel.Error, $"An error has occurred while executing {handlerType}.OnAfterReset during a client reset: {ex}");	6✔
401
402	var exHandle = GCHandle.Alloc(ex);	6✔
403	return GCHandle.ToIntPtr(exHandle);	6✔
404	}
405	}	31✔
406
407	[MonoPInvokeCallback(typeof(NativeMethods.SessionProgressCallback))]
408	private static void HandleSessionProgress(IntPtr tokenPtr, ulong transferredBytes, ulong transferableBytes, double progressEstimate)
409	{
410	var token = (ProgressNotificationToken?)GCHandle.FromIntPtr(tokenPtr).Target;	21✔
411
412	// This is used to provide a reasonable progress estimate until the core work is done
413	double managedProgressEstimate = transferableBytes > 0.0 ? transferredBytes / transferableBytes : 1.0;	21!
414	token?.Notify(managedProgressEstimate);	21!
415	}	21✔
416
417	[MonoPInvokeCallback(typeof(NativeMethods.SessionWaitCallback))]
418	private static void HandleSessionWaitCallback(IntPtr taskCompletionSource, int error_code, PrimitiveValue message)
419	{
420	var handle = GCHandle.FromIntPtr(taskCompletionSource);	565✔
421	var tcs = (TaskCompletionSource)handle.Target!;	565✔
422
423	if (error_code == 0)	565!
424	{
425	tcs.TrySetResult();	565✔
426	}
427	else
428	{
UNCOV 429	var inner = new SessionException(message.AsString(), (ErrorCode)error_code);	×
430	const string OuterMessage = "A system error occurred while waiting for completion. See InnerException for more details";
UNCOV 431	tcs.TrySetException(new RealmException(OuterMessage, inner));	×
432	}
433
434	handle.Free();	565✔
435	}	565✔
436
437	[MonoPInvokeCallback(typeof(NativeMethods.SessionPropertyChangedCallback))]
438	private static void HandleSessionPropertyChangedCallback(IntPtr managedSessionHandle, NotifiableProperty property)
439	{
440	try
441	{
442	if (managedSessionHandle == IntPtr.Zero)	9!
443	{
UNCOV 444	return;	×
445	}
446
447	var propertyName = property switch	9!
448	{	9✔
449	NotifiableProperty.ConnectionState => nameof(Session.ConnectionState),	9✔
UNCOV 450	_ => throw new NotSupportedException($"Unexpected notifiable property value: {property}")	×
451	};	9✔
452	var session = (Session?)GCHandle.FromIntPtr(managedSessionHandle).Target;	9✔
453	if (session is null)	9!
454	{
455	// We're taking a weak handle to the session, so it's possible that it's been collected
UNCOV 456	return;	×
457	}
458
459	ThreadPool.QueueUserWorkItem(_ =>	9✔
460	{	9✔
461	session.RaisePropertyChanged(propertyName);	9✔
462	});	18✔
463	}	9✔
464	catch (Exception ex)	×
465	{
UNCOV 466	Logger.Default.Log(LogLevel.Error, $"An error has occurred while raising a property changed event: {ex}");	×
UNCOV 467	}	×
468	}	9✔
469
470	private enum NotifiableProperty : byte
471	{
472	ConnectionState = 0
473	}
474	}
475	}

realm / realm-dotnet / 9875518284

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous