Enable nullable reference types in Zebus

src/Abc.Zebus.Contracts/Abc.Zebus.Contracts.csproj

@@ -5,6 +5,7 @@
     <PackageId>Zebus.Contracts</PackageId>
     <RootNamespace>Abc.Zebus</RootNamespace>
     <Version>$(ZebusContractsVersion)</Version>
+    <Nullable>enable</Nullable>
   </PropertyGroup>
 
 </Project>

src/Abc.Zebus/Abc.Zebus.csproj

@@ -3,6 +3,7 @@
     <TargetFramework>netstandard2.0</TargetFramework>
     <PackageId>Zebus</PackageId>
     <Version>$(ZebusVersion)</Version>
+    <Nullable>enable</Nullable>
   </PropertyGroup>
 
   <ItemGroup>

src/Abc.Zebus/BusExtensions.cs

@@ -39,7 +39,7 @@ namespace Abc.Zebus
             return Task.WhenAll(sendTasks);
         }
 
-        public static Task<bool> SendMany(this IBus bus, IEnumerable<ICommand> commands, Action<ICommand, CommandResult> onCommandExecuted = null)
+        public static Task<bool> SendMany(this IBus bus, IEnumerable<ICommand> commands, Action<ICommand, CommandResult>? onCommandExecuted = null)
         {
             var sendTasks = commands.Select(command =>
             {

src/Abc.Zebus/CommandResult.cs

@@ -8,12 +8,12 @@ namespace Abc.Zebus
     public class CommandResult
     {
         [Obsolete("Use the constructor with the responseMessage parameter")]
-        public CommandResult(int errorCode, object response)
+        public CommandResult(int errorCode, object? response)
             : this(errorCode, null, response)
         {
         }
 
-        public CommandResult(int errorCode, string responseMessage, object response)
+        public CommandResult(int errorCode, string? responseMessage, object? response)
         {
             ErrorCode = errorCode;
             ResponseMessage = responseMessage;
@@ -21,8 +21,8 @@ namespace Abc.Zebus
         }
 
         public int ErrorCode { get; }
-        public string ResponseMessage { get; }
-        public object Response { get; }
+        public string? ResponseMessage { get; }
+        public object? Response { get; }
 
         public bool IsSuccess => ErrorCode == 0;
 

src/Abc.Zebus/Core/Bus.cs

@@ -14,7 +14,6 @@ using Abc.Zebus.Serialization;
 using Abc.Zebus.Transport;
 using Abc.Zebus.Util;
 using Abc.Zebus.Util.Extensions;
-using JetBrains.Annotations;
 using log4net;
 using Newtonsoft.Json;
 
@@ -38,8 +37,7 @@ namespace Abc.Zebus.Core
         private readonly IBindingKeyPredicateBuilder _predicateBuilder;
         private readonly IBusConfiguration _configuration;
 
-        [CanBeNull]
-        private Task _processPendingUnsubscriptionsTask;
+        private Task? _processPendingUnsubscriptionsTask;
 
         private int _subscriptionsVersion;
         private int _status;
@@ -58,13 +56,13 @@ namespace Abc.Zebus.Core
             _configuration = configuration;
         }
 
-        public event Action Starting;
-        public event Action Started;
-        public event Action Stopping;
-        public event Action Stopped;
+        public event Action? Starting;
+        public event Action? Started;
+        public event Action? Stopping;
+        public event Action? Stopped;
 
         public PeerId PeerId { get; private set; }
-        public string Environment { get; private set; }
+        public string Environment { get; private set; } = string.Empty;
         public bool IsRunning => Status == BusStatus.Started || Status == BusStatus.Stopping;
         public string EndPoint => _transport.InboundEndPoint;
         public string DeserializationFailureDumpDirectoryPath { get; set; } = PathUtil.InBaseDirectory("deserialization_failure_dumps");
@@ -284,7 +282,7 @@ namespace Abc.Zebus.Core
             return taskCompletionSource.Task;
         }
 
-        public async Task<IDisposable> SubscribeAsync([NotNull] SubscriptionRequest request)
+        public async Task<IDisposable> SubscribeAsync(SubscriptionRequest request)
         {
             ValidateSubscriptionRequest(request);
 
@@ -301,7 +299,7 @@ namespace Abc.Zebus.Core
             return new DisposableAction(() => RemoveSubscriptions(request));
         }
 
-        public async Task<IDisposable> SubscribeAsync([NotNull] SubscriptionRequest request, [NotNull] Action<IMessage> handler)
+        public async Task<IDisposable> SubscribeAsync(SubscriptionRequest request, Action<IMessage> handler)
         {
             ValidateSubscriptionRequest(request);
 
@@ -312,7 +310,12 @@ namespace Abc.Zebus.Core
 
             var eventHandlerInvokers = request.Subscriptions
                                               .GroupBy(x => x.MessageTypeId)
-                                              .Select(x => new DynamicMessageHandlerInvoker(handler, x.Key.GetMessageType(), x.Select(s => s.BindingKey).ToList(), _predicateBuilder))
+                                              .Select(x => new DynamicMessageHandlerInvoker(
+                                                          handler,
+                                                          x.Key.GetMessageType() ?? throw new InvalidOperationException($"Could not resolve type {x.Key.FullName}"),
+                                                          x.Select(s => s.BindingKey).ToList(),
+                                                          _predicateBuilder
+                                                      ))
                                               .ToList();
 
             if (request.Batch != null)
@@ -410,7 +413,7 @@ namespace Abc.Zebus.Core
 
         internal async Task WhenUnsubscribeCompletedAsync()
         {
-            Task task;
+            Task? task;
 
             lock (_subscriptions)
             {
@@ -513,7 +516,7 @@ namespace Abc.Zebus.Core
         public void Reply(int errorCode)
             => Reply(errorCode, null);
 
-        public void Reply(int errorCode, string message)
+        public void Reply(int errorCode, string? message)
         {
             var messageContext = MessageContext.Current;
             if (messageContext == null)
@@ -523,7 +526,7 @@ namespace Abc.Zebus.Core
             messageContext.ReplyMessage = message;
         }
 
-        public void Reply(IMessage response)
+        public void Reply(IMessage? response)
         {
             var messageContext = MessageContext.Current;
             if (messageContext == null)
@@ -548,10 +551,10 @@ namespace Abc.Zebus.Core
             }
         }
 
-        public MessageDispatch CreateMessageDispatch(TransportMessage transportMessage)
+        public MessageDispatch? CreateMessageDispatch(TransportMessage transportMessage)
             => CreateMessageDispatch(transportMessage, synchronousDispatch: false, sendAcknowledgment: false);
 
-        private MessageDispatch CreateMessageDispatch(TransportMessage transportMessage, bool synchronousDispatch, bool sendAcknowledgment = true)
+        private MessageDispatch? CreateMessageDispatch(TransportMessage transportMessage, bool synchronousDispatch, bool sendAcknowledgment = true)
         {
             var message = ToMessage(transportMessage);
             if (message == null)
@@ -596,7 +599,7 @@ namespace Abc.Zebus.Core
             };
         }
 
-        private void HandleDispatchErrors(MessageDispatch dispatch, DispatchResult dispatchResult, TransportMessage failingTransportMessage = null)
+        private void HandleDispatchErrors(MessageDispatch dispatch, DispatchResult dispatchResult, TransportMessage? failingTransportMessage = null)
         {
             if (!_configuration.IsErrorPublicationEnabled || !IsRunning || dispatchResult.Errors.Count == 0 || dispatchResult.Errors.All(error => error is DomainException))
                 return;
@@ -644,7 +647,7 @@ namespace Abc.Zebus.Core
 
         private void HandleMessageExecutionCompleted(TransportMessage transportMessage)
         {
-            var message = (MessageExecutionCompleted)ToMessage(transportMessage);
+            var message = (MessageExecutionCompleted?)ToMessage(transportMessage);
             if (message == null)
                 return;
 
@@ -664,7 +667,7 @@ namespace Abc.Zebus.Core
             taskCompletionSource.SetResult(commandResult);
         }
 
-        protected virtual void HandleLocalMessage(IMessage message, TaskCompletionSource<CommandResult> taskCompletionSource)
+        protected virtual void HandleLocalMessage(IMessage message, TaskCompletionSource<CommandResult>? taskCompletionSource)
         {
             _messageLogger.LogReceiveMessageLocal(message);
 
@@ -677,7 +680,7 @@ namespace Abc.Zebus.Core
             _messageDispatcher.Dispatch(dispatch);
         }
 
-        private Action<MessageDispatch, DispatchResult> GetOnLocalMessageDispatchedContinuation(TaskCompletionSource<CommandResult> taskCompletionSource)
+        private Action<MessageDispatch, DispatchResult> GetOnLocalMessageDispatchedContinuation(TaskCompletionSource<CommandResult>? taskCompletionSource)
         {
             return (dispatch, dispatchResult) =>
             {
@@ -719,11 +722,14 @@ namespace Abc.Zebus.Core
         protected TransportMessage ToTransportMessage(IMessage message)
             => _serializer.ToTransportMessage(message, PeerId, EndPoint);
 
-        private IMessage ToMessage(TransportMessage transportMessage)
+        private IMessage? ToMessage(TransportMessage transportMessage)
             => ToMessage(transportMessage.MessageTypeId, transportMessage.Content, transportMessage);
 
-        private IMessage ToMessage(MessageTypeId messageTypeId, Stream messageStream, TransportMessage transportMessage)
+        private IMessage? ToMessage(MessageTypeId messageTypeId, Stream? messageStream, TransportMessage transportMessage)
         {
+            if (messageStream is null)
+                return null;
+
             try
             {
                 return _serializer.ToMessage(transportMessage, messageTypeId, messageStream);

src/Abc.Zebus/Core/BusFactory.cs

@@ -16,8 +16,8 @@ namespace Abc.Zebus.Core
         private readonly List<Action<ConfigurationExpression>> _configurationActions = new List<Action<ConfigurationExpression>>();
         private readonly ZmqTransportConfiguration _transportConfiguration = new ZmqTransportConfiguration();
         private readonly List<ScanTarget> _scanTargets = new List<ScanTarget>();
-        private IBusConfiguration _configuration;
-        private string _environment;
+        private IBusConfiguration? _configuration;
+        private string? _environment;
         public PeerId PeerId { get; set; }
 
         public BusFactory()
@@ -97,8 +97,9 @@ namespace Abc.Zebus.Core
 
         public IBus CreateBus()
         {
-            if (_configuration == null)
+            if (_configuration == null || _environment == null)
                 throw new InvalidOperationException("The CreateBus() method was called with no configuration (Call .WithConfiguration(...) first)");
+
             Container.Configure(x => x.AddRegistry<ZebusRegistry>());
             Container.Configure(x =>
             {
@@ -170,10 +171,10 @@ namespace Abc.Zebus.Core
 
         private class ScanTarget
         {
-            private readonly Assembly _assembly;
-            private readonly Type _type;
+            private readonly Assembly? _assembly;
+            private readonly Type? _type;
 
-            public ScanTarget(Assembly assembly, Type type)
+            public ScanTarget(Assembly? assembly, Type? type)
             {
                 _assembly = assembly;
                 _type = type;

src/Abc.Zebus/Core/BusMessageLogger.cs

@@ -1,6 +1,7 @@
 using System;
 using System.Collections.Concurrent;
 using System.Collections.Generic;
+using System.Diagnostics.CodeAnalysis;
 using System.Reflection;
 using Abc.Zebus.Scan;
 using Abc.Zebus.Transport;
@@ -41,7 +42,7 @@ namespace Abc.Zebus.Core
         public bool IsInfoEnabled(IMessage message)
             => _logInfoEnabled && GetLogHelper(message).Logger.IsInfoEnabled;
 
-        public void LogHandleMessage(IList<IMessage> messages, string dispatchQueueName, MessageId? messageId)
+        public void LogHandleMessage(IList<IMessage> messages, string? dispatchQueueName, MessageId? messageId)
         {
             var message = messages[0];
 
@@ -80,7 +81,7 @@ namespace Abc.Zebus.Core
                 return;
 
             var messageText = logHelper.GetMessageText(message);
-            _logger.Debug($"RECV remote: {messageText} from {transportMessage.SenderId} ({transportMessage.Content.Length} bytes). [{transportMessage.Id}]");
+            _logger.Debug($"RECV remote: {messageText} from {transportMessage.SenderId} ({transportMessage.Content?.Length} bytes). [{transportMessage.Id}]");
         }
 
         public void LogSendMessage(IMessage message, IList<Peer> peers)
@@ -102,7 +103,7 @@ namespace Abc.Zebus.Core
             var messageText = logHelper.GetMessageText(message);
             var targetPeersText = GetTargetPeersText(peers);
 
-            _logger.Info($"SEND: {messageText} to {targetPeersText} ({transportMessage.Content.Length} bytes) [{transportMessage.Id}]");
+            _logger.Info($"SEND: {messageText} to {targetPeersText} ({transportMessage.Content?.Length} bytes) [{transportMessage.Id}]");
         }
 
         private static string GetTargetPeersText(IList<Peer> peers)
@@ -126,6 +127,7 @@ namespace Abc.Zebus.Core
         public static string ToString(IMessage message)
             => GetLogHelper(message).GetMessageText(message);
 
+        [SuppressMessage("ReSharper", "ConvertClosureToMethodGroup")]
         private static MessageTypeLogHelper GetLogHelper(IMessage message)
             => _logHelpers.GetOrAdd(message.GetType(), type => CreateLogger(type));
 
@@ -137,7 +139,7 @@ namespace Abc.Zebus.Core
             return new MessageTypeLogHelper(logger, hasToStringOverride, messageType.GetPrettyName());
         }
 
-        private bool TryGetLogHelperForInfo(IMessage message, out MessageTypeLogHelper logHelper)
+        private bool TryGetLogHelperForInfo(IMessage message, [NotNullWhen(true)] out MessageTypeLogHelper? logHelper)
         {
             if (!_logInfoEnabled)
             {
@@ -149,7 +151,7 @@ namespace Abc.Zebus.Core
             return logHelper.Logger.IsInfoEnabled;
         }
 
-        private bool TryGetLogHelperForDebug(IMessage message, out MessageTypeLogHelper logHelper)
+        private bool TryGetLogHelperForDebug(IMessage message, [NotNullWhen(true)] out MessageTypeLogHelper? logHelper)
         {
             if (!_logDebugEnabled)
             {

src/Abc.Zebus/Core/MessageContextAwareBus.cs

@@ -53,13 +53,13 @@ namespace Abc.Zebus.Core
 
         public void Reply(int errorCode) => Reply(errorCode, null);
 
-        public void Reply(int errorCode, string message)
+        public void Reply(int errorCode, string? message)
         {
             MessageContext.ReplyCode = errorCode;
             MessageContext.ReplyMessage = message;
         }
 
-        public void Reply(IMessage response)
+        public void Reply(IMessage? response)
         {
             MessageContext.ReplyResponse = response;
         }
@@ -67,25 +67,25 @@ namespace Abc.Zebus.Core
         public void Start() => _bus.Start();
         public void Stop() => _bus.Stop();
 
-        public event Action Starting
+        public event Action? Starting
         {
             add => _bus.Starting += value;
             remove => _bus.Starting -= value;
         }
 
-        public event Action Started
+        public event Action? Started
         {
             add => _bus.Started += value;
             remove => _bus.Started -= value;
         }
 
-        public event Action Stopping
+        public event Action? Stopping
         {
             add => _bus.Stopping += value;
             remove => _bus.Stopping -= value;
         }
 
-        public event Action Stopped
+        public event Action? Stopped
         {
             add => _bus.Stopped += value;
             remove => _bus.Stopped -= value;

src/Abc.Zebus/Core/MessageExecutionCompleted.cs

@@ -1,5 +1,6 @@
 using System;
 using System.Collections.Generic;
+using System.Diagnostics.CodeAnalysis;
 using System.IO;
 using System.Linq;
 using Abc.Zebus.Dispatch;
@@ -9,6 +10,7 @@ using ProtoBuf;
 namespace Abc.Zebus.Core
 {
     [ProtoContract, Transient, Infrastructure]
+    [SuppressMessage("ReSharper", "AutoPropertyCanBeMadeGetOnly.Local")]
     public class MessageExecutionCompleted : IMessage
     {
         public static readonly MessageTypeId TypeId = new MessageTypeId(typeof(MessageExecutionCompleted));
@@ -23,10 +25,10 @@ namespace Abc.Zebus.Core
         public MessageTypeId? PayloadTypeId { get; private set; }
 
         [ProtoMember(4, IsRequired = false)]
-        public byte[] Payload { get; private set; }
+        public byte[]? Payload { get; private set; }
 
         [ProtoMember(5, IsRequired = false)]
-        public string ResponseMessage { get; private set; } = string.Empty;
+        public string? ResponseMessage { get; private set; } = string.Empty;
 
         [Obsolete("Use the constructor with the responseMessage parameter")]
         public MessageExecutionCompleted(MessageId sourceCommandId, int errorCode)
@@ -34,14 +36,14 @@ namespace Abc.Zebus.Core
         {
         }
 
-        public MessageExecutionCompleted(MessageId sourceCommandId, int errorCode, string responseMessage)
+        public MessageExecutionCompleted(MessageId sourceCommandId, int errorCode, string? responseMessage)
         {
             SourceCommandId = sourceCommandId;
             ErrorCode = errorCode;
             ResponseMessage = responseMessage ?? string.Empty;
         }
 
-        public MessageExecutionCompleted(MessageId sourceCommandId, MessageTypeId payloadTypeId, byte[] payload)
+        public MessageExecutionCompleted(MessageId sourceCommandId, MessageTypeId payloadTypeId, byte[]? payload)
         {
             SourceCommandId = sourceCommandId;
             ErrorCode = 0;

src/Abc.Zebus/Core/RoundRobinPeerSelector.cs

@@ -8,7 +8,7 @@ namespace Abc.Zebus.Core
     {
         private readonly ConcurrentDictionary<Type, int> _peerIndexes = new ConcurrentDictionary<Type, int>();
 
-        public Peer GetTargetPeer(ICommand command, IList<Peer> handlingPeers)
+        public Peer? GetTargetPeer(ICommand command, IList<Peer> handlingPeers)
         {
             if (handlingPeers.Count == 1)
                 return handlingPeers[0];
@@ -18,8 +18,7 @@ namespace Abc.Zebus.Core
 
             var commandType = command.GetType();
 
-            int index;
-            if (!_peerIndexes.TryGetValue(commandType, out index))
+            if (!_peerIndexes.TryGetValue(commandType, out var index))
                 index = 0;
 
             if (index >= handlingPeers.Count)
@@ -32,4 +31,4 @@ namespace Abc.Zebus.Core
             return resolvedPeer;
         }
     }
-}
+}

src/Abc.Zebus/Directory/IPeerDirectory.cs

@@ -20,7 +20,7 @@ namespace Abc.Zebus.Directory
         bool IsPersistent(PeerId peerId);
 
         // TODO: move to a specific interface (IPeerDirectoryExplorer)
-        PeerDescriptor GetPeerDescriptor(PeerId peerId);
+        PeerDescriptor? GetPeerDescriptor(PeerId peerId);
 
         IEnumerable<PeerDescriptor> GetPeerDescriptors();
     }

src/Abc.Zebus/Directory/MessageBinding.cs

@@ -24,7 +24,7 @@ namespace Abc.Zebus.Directory
         public bool Equals(MessageBinding other)
             => MessageTypeId == other.MessageTypeId && RoutingKey.Equals(other.RoutingKey);
 
-        public override bool Equals(object obj)
+        public override bool Equals(object? obj)
             => obj is MessageBinding binding && Equals(binding);
 
         public override int GetHashCode()

src/Abc.Zebus/Directory/PeerDescriptor.cs

@@ -43,6 +43,8 @@ namespace Abc.Zebus.Directory
         [UsedImplicitly]
         private PeerDescriptor()
         {
+            Peer = default!;
+            Subscriptions = Array.Empty<Subscription>();
         }
 
         public PeerId PeerId => Peer.Id;

src/Abc.Zebus/Directory/PeerDirectoryClient.PeerEntry.cs

@@ -20,11 +20,11 @@ namespace Abc.Zebus.Directory
                 IsPersistent = descriptor.IsPersistent;
                 TimestampUtc = descriptor.TimestampUtc ?? DateTime.UtcNow;
                 HasDebuggerAttached = descriptor.HasDebuggerAttached;
-                
+
                 _globalSubscriptionsIndex = globalSubscriptionsIndex;
             }
 
-            public Peer Peer { get; private set; }
+            public Peer Peer { get; }
             public bool IsPersistent { get; set; }
             public DateTime TimestampUtc { get; set; }
             public bool HasDebuggerAttached { get; set; }
@@ -76,7 +76,7 @@ namespace Abc.Zebus.Directory
             private void SetSubscriptionsForType(MessageTypeId messageTypeId, IEnumerable<BindingKey> bindingKeys, DateTime? timestampUtc)
             {
                 var newBindingKeys = bindingKeys.ToHashSet();
-                
+
                 var messageTypeEntry = _peerSubscriptionsByMessageType.GetValueOrAdd(messageTypeId, MessageTypeEntry.Create);
                 if (messageTypeEntry.TimestampUtc > timestampUtc)
                     return;
@@ -144,4 +144,4 @@ namespace Abc.Zebus.Directory
             }
         }
     }
-}
+}

src/Abc.Zebus/Directory/PeerDirectoryClient.cs

@@ -27,16 +27,16 @@ namespace Abc.Zebus.Directory
         private readonly UniqueTimestampProvider _timestampProvider = new UniqueTimestampProvider(10);
         private readonly IBusConfiguration _configuration;
         private readonly Stopwatch _pingStopwatch = new Stopwatch();
-        private BlockingCollection<IEvent> _messagesReceivedDuringRegister;
-        private IEnumerable<Peer> _directoryPeers;
-        private Peer _self;
+        private BlockingCollection<IEvent>? _messagesReceivedDuringRegister;
+        private IEnumerable<Peer> _directoryPeers = Enumerable.Empty<Peer>();
+        private Peer _self = default!;
 
         public PeerDirectoryClient(IBusConfiguration configuration)
         {
             _configuration = configuration;
         }
 
-        public event Action<PeerId, PeerUpdateAction> PeerUpdated;
+        public event Action<PeerId, PeerUpdateAction>? PeerUpdated;
 
         public TimeSpan TimeSinceLastPing => _pingStopwatch.IsRunning ? _pingStopwatch.Elapsed : TimeSpan.MaxValue;
 
@@ -67,7 +67,7 @@ namespace Abc.Zebus.Directory
 
         private void ProcessMessagesReceivedDuringRegister()
         {
-            foreach (var message in _messagesReceivedDuringRegister.GetConsumingEnumerable())
+            foreach (var message in _messagesReceivedDuringRegister!.GetConsumingEnumerable())
             {
                 try
                 {
@@ -146,7 +146,7 @@ namespace Abc.Zebus.Directory
             {
                 var registration = await bus.Send(new RegisterPeerCommand(self), directoryPeer).WithTimeoutAsync(_configuration.RegistrationTimeout).ConfigureAwait(false);
 
-                var response = (RegisterPeerResponse)registration.Response;
+                var response = (RegisterPeerResponse?)registration.Response;
                 if (response?.PeerDescriptors == null)
                     return false;
 
@@ -231,7 +231,7 @@ namespace Abc.Zebus.Directory
             return entry != null && entry.IsPersistent;
         }
 
-        public PeerDescriptor GetPeerDescriptor(PeerId peerId)
+        public PeerDescriptor? GetPeerDescriptor(PeerId peerId)
             => _peers.GetValueOrDefault(peerId)?.ToPeerDescriptor();
 
         public IEnumerable<PeerDescriptor> GetPeerDescriptors()

src/Abc.Zebus/Directory/PeerStopped.cs

@@ -11,7 +11,7 @@ namespace Abc.Zebus.Directory
         public readonly PeerId PeerId;
 
         [ProtoMember(2, IsRequired = false)]
-        public readonly string PeerEndPoint;
+        public readonly string? PeerEndPoint;
 
         [ProtoMember(3, IsRequired = false)]
         public readonly DateTime? TimestampUtc;
@@ -21,7 +21,7 @@ namespace Abc.Zebus.Directory
         {
         }
 
-        public PeerStopped(PeerId peerId, string peerEndPoint, DateTime? timestampUtc = null)
+        public PeerStopped(PeerId peerId, string? peerEndPoint, DateTime? timestampUtc = null)
         {
             PeerId = peerId;
             PeerEndPoint = peerEndPoint;
@@ -30,4 +30,4 @@ namespace Abc.Zebus.Directory
 
         public override string ToString() => $"{PeerId} TimestampUtc: {TimestampUtc:yyyy-MM-dd HH:mm:ss.fff}";
     }
-}
+}

src/Abc.Zebus/Directory/PeerSubscriptionTree.cs

@@ -5,7 +5,6 @@ using System.Diagnostics.CodeAnalysis;
 using System.Linq;
 using Abc.Zebus.Routing;
 using Abc.Zebus.Util.Extensions;
-using JetBrains.Annotations;
 
 namespace Abc.Zebus.Directory
 {
@@ -101,17 +100,12 @@ namespace Abc.Zebus.Directory
         private class SubscriptionNode
         {
             private static readonly Action<SubscriptionNode, string> _removeNode = (x, part) => x.RemoveChildNode(part);
-            private static readonly Action<SubscriptionNode, string> _removeSharpNode = (x, _) => x._sharpNode = null;
-            private static readonly Action<SubscriptionNode, string> _removeStarNode = (x, _) => x._starNode = null;
+            private static readonly Action<SubscriptionNode, string?> _removeSharpNode = (x, _) => x._sharpNode = null;
+            private static readonly Action<SubscriptionNode, string?> _removeStarNode = (x, _) => x._starNode = null;
 
-            [CanBeNull]
-            private ConcurrentDictionary<string, SubscriptionNode> _childNodes;
-
-            [CanBeNull]
-            private SubscriptionNode _sharpNode;
-
-            [CanBeNull]
-            private SubscriptionNode _starNode;
+            private ConcurrentDictionary<string, SubscriptionNode>? _childNodes;
+            private SubscriptionNode? _sharpNode;
+            private SubscriptionNode? _starNode;
 
             private readonly int _nextPartIndex;
             private List<Peer> _peers = new List<Peer>();
@@ -178,10 +172,10 @@ namespace Abc.Zebus.Directory
                     return UpdateChildNode(GetOrCreateStarNode(), peer, subscription, action, null, _removeStarNode);
 
                 var childNode = GetOrAddChildNode(nextPart);
-                return UpdateChildNode(childNode, peer, subscription, action, nextPart, _removeNode);
+                return UpdateChildNode(childNode, peer, subscription, action, nextPart, _removeNode!);
             }
 
-            private int UpdateChildNode(SubscriptionNode childNode, Peer peer, BindingKey subscription, UpdateAction action, string childNodePart, Action<SubscriptionNode, string> remover)
+            private int UpdateChildNode(SubscriptionNode childNode, Peer peer, BindingKey subscription, UpdateAction action, string? childNodePart, Action<SubscriptionNode, string?> remover)
             {
                 var update = childNode.Update(peer, subscription, action);
                 _peerCountIncludingChildren += update;
@@ -212,10 +206,10 @@ namespace Abc.Zebus.Directory
                 => _childNodes?.TryRemove(part, out _);
 
             private SubscriptionNode GetOrCreateSharpNode()
-                => _sharpNode ?? (_sharpNode = new SubscriptionNode(_nextPartIndex + 1));
+                => _sharpNode ??= new SubscriptionNode(_nextPartIndex + 1);
 
             private SubscriptionNode GetOrCreateStarNode()
-                => _starNode ?? (_starNode = new SubscriptionNode(_nextPartIndex + 1));
+                => _starNode ??= new SubscriptionNode(_nextPartIndex + 1);
 
             private int UpdateList(Peer peer, UpdateAction action)
                 => action == UpdateAction.Add

src/Abc.Zebus/Directory/SubscriptionsForType.cs

@@ -14,7 +14,7 @@ namespace Abc.Zebus.Directory
         public readonly MessageTypeId MessageTypeId;
 
         [ProtoMember(2, IsRequired = true)]
-        public readonly BindingKey[] BindingKeys = Array.Empty<BindingKey>();
+        public readonly BindingKey[] BindingKeys;
 
         public SubscriptionsForType(MessageTypeId messageTypeId, params BindingKey[] bindingKeys)
         {
@@ -25,6 +25,7 @@ namespace Abc.Zebus.Directory
         [UsedImplicitly]
         private SubscriptionsForType()
         {
+            BindingKeys = Array.Empty<BindingKey>();
         }
 
         public static SubscriptionsForType Create<TMessage>(params BindingKey[] bindingKeys)
@@ -36,12 +37,12 @@ namespace Abc.Zebus.Directory
                             .ToDictionary(grp => grp.Key, grp => new SubscriptionsForType(grp.Key, grp.Select(sub => sub.BindingKey).ToArray()));
 
         public Subscription[] ToSubscriptions()
-            => BindingKeys?.Select(bindingKey => new Subscription(MessageTypeId, bindingKey)).ToArray() ?? new Subscription[0];
+            => BindingKeys?.Select(bindingKey => new Subscription(MessageTypeId, bindingKey)).ToArray() ?? Array.Empty<Subscription>();
 
-        public bool Equals(SubscriptionsForType other)
+        public bool Equals(SubscriptionsForType? other)
             => other != null && MessageTypeId == other.MessageTypeId && BindingKeys.SequenceEqual(other.BindingKeys);
 
-        public override bool Equals(object obj)
+        public override bool Equals(object? obj)
         {
             if (ReferenceEquals(null, obj))
                 return false;

src/Abc.Zebus/Directory/UnregisterPeerCommand.cs

@@ -11,7 +11,7 @@ namespace Abc.Zebus.Directory
         public readonly PeerId PeerId;
 
         [ProtoMember(2, IsRequired = false)]
-        public readonly string PeerEndPoint;
+        public readonly string? PeerEndPoint;
 
         [ProtoMember(3, IsRequired = false)]
         public readonly DateTime? TimestampUtc;
@@ -21,7 +21,7 @@ namespace Abc.Zebus.Directory
         {
         }
 
-        public UnregisterPeerCommand(PeerId peerId, string peerEndPoint, DateTime? timestampUtc = null)
+        public UnregisterPeerCommand(PeerId peerId, string? peerEndPoint, DateTime? timestampUtc = null)
         {
             PeerId = peerId;
             PeerEndPoint = peerEndPoint;
@@ -30,4 +30,4 @@ namespace Abc.Zebus.Directory
 
         public override string ToString() => $"{PeerId} TimestampUtc: {TimestampUtc:yyyy-MM-dd HH:mm:ss.fff}";
     }
-}
+}

src/Abc.Zebus/Dispatch/BatchedMessageHandlerInvoker.cs

@@ -33,8 +33,9 @@ namespace Abc.Zebus.Dispatch
 
         public override bool CanMergeWith(IMessageHandlerInvoker other)
         {
-            var otherBatchedInvoker = other as BatchedMessageHandlerInvoker;
-            return otherBatchedInvoker != null && otherBatchedInvoker.MessageHandlerType == MessageHandlerType && otherBatchedInvoker.MessageType == MessageType;
+            return other is BatchedMessageHandlerInvoker otherBatchedInvoker
+                   && otherBatchedInvoker.MessageHandlerType == MessageHandlerType
+                   && otherBatchedInvoker.MessageType == MessageType;
         }
 
         private object CreateHandler(MessageContext messageContext)

src/Abc.Zebus/Dispatch/DispatchQueue.cs

@@ -12,14 +12,14 @@ namespace Abc.Zebus.Dispatch
     public class DispatchQueue : IDisposable
     {
         [ThreadStatic]
-        private static string _currentDispatchQueueName;
+        private static string? _currentDispatchQueueName;
 
         private static readonly ILog _logger = LogManager.GetLogger(typeof(DispatchQueue));
 
         private readonly IPipeManager _pipeManager;
         private readonly int _batchSize;
         private FlushableBlockingCollection<Entry> _queue = new FlushableBlockingCollection<Entry>();
-        private Thread _thread;
+        private Thread? _thread;
         private volatile bool _isRunning;
         private int _asyncInvocationsCount;
         private int _hasCompletedAsyncInvocationsSinceLastWait;
@@ -85,7 +85,7 @@ namespace Abc.Zebus.Dispatch
 
             _queue.CompleteAdding();
 
-            _thread.Join();
+            _thread?.Join();
 
             _queue = new FlushableBlockingCollection<Entry>();
         }
@@ -165,12 +165,12 @@ namespace Abc.Zebus.Dispatch
                 if (!_isRunning)
                     return;
 
-                switch (batch.FirstEntry.Invoker.Mode)
+                switch (batch.FirstEntry.Invoker!.Mode)
                 {
                     case MessageHandlerInvokerMode.Synchronous:
                     {
                         SynchronizationContext.SetSynchronizationContext(null);
-                        var invocation = _pipeManager.BuildPipeInvocation(batch.FirstEntry.Invoker, batch.Messages, batch.FirstEntry.Dispatch.Context);
+                        var invocation = _pipeManager.BuildPipeInvocation(batch.FirstEntry.Invoker, batch.Messages, batch.FirstEntry.Dispatch!.Context);
                         invocation.Run();
                         batch.SetHandled(null);
                         break;
@@ -180,7 +180,7 @@ namespace Abc.Zebus.Dispatch
                     {
                         SynchronizationContext.SetSynchronizationContext(SynchronizationContext);
                         var asyncBatch = batch.Clone();
-                        var invocation = _pipeManager.BuildPipeInvocation(asyncBatch.FirstEntry.Invoker, asyncBatch.Messages, asyncBatch.FirstEntry.Dispatch.Context);
+                        var invocation = _pipeManager.BuildPipeInvocation(asyncBatch.FirstEntry.Invoker!, asyncBatch.Messages, asyncBatch.FirstEntry.Dispatch!.Context);
                         Interlocked.Increment(ref _asyncInvocationsCount);
                         invocation.RunAsync().ContinueWith(task => OnAsyncBatchCompleted(task, asyncBatch), TaskContinuationOptions.ExecuteSynchronously);
                         break;
@@ -277,7 +277,7 @@ namespace Abc.Zebus.Dispatch
         }
 
         // for unit tests
-        internal static string GetCurrentDispatchQueueName()
+        internal static string? GetCurrentDispatchQueueName()
         {
             return _currentDispatchQueueName;
         }
@@ -306,9 +306,9 @@ namespace Abc.Zebus.Dispatch
                 Action = action;
             }
 
-            public readonly MessageDispatch Dispatch;
-            public readonly IMessageHandlerInvoker Invoker;
-            public readonly Action Action;
+            public readonly MessageDispatch? Dispatch;
+            public readonly IMessageHandlerInvoker? Invoker;
+            public readonly Action? Action;
         }
 
         private class Batch
@@ -328,20 +328,20 @@ namespace Abc.Zebus.Dispatch
             public void Add(Entry entry)
             {
                 Entries.Add(entry);
-                Messages.Add(entry.Dispatch.Message);
+                Messages.Add(entry.Dispatch!.Message);
             }
 
-            public void SetHandled(Exception error)
+            public void SetHandled(Exception? error)
             {
                 foreach (var entry in Entries)
                 {
                     try
                     {
-                        entry.Dispatch.SetHandled(entry.Invoker, error);
+                        entry.Dispatch!.SetHandled(entry.Invoker!, error);
                     }
                     catch (Exception ex)
                     {
-                        _logger.Error($"Unable to run dispatch continuation, MessageType: {entry.Invoker.MessageType.Name}, HandlerType: {entry.Invoker.MessageHandlerType.Name}, HandlerError: {error}, ContinuationError: {ex}");
+                        _logger.Error($"Unable to run dispatch continuation, MessageType: {entry.Invoker!.MessageType?.Name}, HandlerType: {entry.Invoker!.MessageHandlerType.Name}, HandlerError: {error}, ContinuationError: {ex}");
                     }
                 }
             }
@@ -368,7 +368,7 @@ namespace Abc.Zebus.Dispatch
                 if (IsEmpty)
                     return true;
 
-                return entry.Invoker.CanMergeWith(FirstEntry.Invoker);
+                return entry.Invoker!.CanMergeWith(FirstEntry.Invoker!);
             }
         }
 

src/Abc.Zebus/Dispatch/DispatchQueueNameAttribute.cs

@@ -10,6 +10,6 @@ namespace Abc.Zebus.Dispatch
             QueueName = queueName;
         }
 
-        public string QueueName { get; private set; }
+        public string QueueName { get; }
     }
-}
+}

src/Abc.Zebus/Dispatch/DispatchResult.cs

@@ -8,11 +8,11 @@ namespace Abc.Zebus.Dispatch
         private static readonly Dictionary<Type, Exception> _noErrors = new Dictionary<Type, Exception>();
         private readonly IDictionary<Type, Exception> _errors;
 
-        public DispatchResult(IDictionary<Type, Exception> errorsByHandlerType)
+        public DispatchResult(IDictionary<Type, Exception>? errorsByHandlerType)
         {
             _errors = errorsByHandlerType ?? _noErrors;
         }
-        
+
         public ICollection<Exception> Errors => _errors.Values;
 
         public ICollection<Type> ErrorHandlerTypes => _errors.Keys;

src/Abc.Zebus/Dispatch/IMessageDispatchFactory.cs

@@ -4,6 +4,6 @@ namespace Abc.Zebus.Dispatch
 {
     public interface IMessageDispatchFactory
     {
-        MessageDispatch CreateMessageDispatch(TransportMessage transportMessage);
+        MessageDispatch? CreateMessageDispatch(TransportMessage transportMessage);
     }
 }

src/Abc.Zebus/Dispatch/IMessageHandlerInvoker.cs

@@ -6,7 +6,7 @@ namespace Abc.Zebus.Dispatch
     public interface IMessageHandlerInvoker
     {
         Type MessageHandlerType { get; }
-        Type MessageType { get; }
+        Type? MessageType { get; }
         MessageTypeId MessageTypeId { get; }
         bool ShouldBeSubscribedOnStartup { get; }
         string DispatchQueueName { get; }
@@ -17,4 +17,4 @@ namespace Abc.Zebus.Dispatch
         bool ShouldHandle(IMessage message);
         bool CanMergeWith(IMessageHandlerInvoker other);
     }
-}
+}

src/Abc.Zebus/Dispatch/MessageDispatch.cs

@@ -11,7 +11,7 @@ namespace Abc.Zebus.Dispatch
 
         private readonly IMessageSerializer _messageSerializer;
         private readonly Action<MessageDispatch, DispatchResult> _continuation;
-        private Dictionary<Type, Exception> _exceptions;
+        private Dictionary<Type, Exception>? _exceptions;
         private int _remainingHandlerCount;
         private bool _isCloned;
 
@@ -35,7 +35,7 @@ namespace Abc.Zebus.Dispatch
             _continuation(this, new DispatchResult(null));
         }
 
-        public void SetHandled(IMessageHandlerInvoker invoker, Exception error)
+        public void SetHandled(IMessageHandlerInvoker invoker, Exception? error)
         {
             if (error != null)
                 AddException(invoker.MessageHandlerType, error);

src/Abc.Zebus/Dispatch/MessageDispatcher.cs

@@ -18,9 +18,9 @@ namespace Abc.Zebus.Dispatch
         private readonly IMessageHandlerInvokerLoader[] _invokerLoaders;
         private readonly IDispatchQueueFactory _dispatchQueueFactory;
         private ConcurrentDictionary<MessageTypeId, List<IMessageHandlerInvoker>> _invokers = new ConcurrentDictionary<MessageTypeId, List<IMessageHandlerInvoker>>();
-        private Func<Assembly, bool> _assemblyFilter;
-        private Func<Type, bool> _handlerFilter;
-        private Func<Type, bool> _messageFilter;
+        private Func<Assembly, bool>? _assemblyFilter;
+        private Func<Type, bool>? _handlerFilter;
+        private Func<Type, bool>? _messageFilter;
         private MessageDispatcherStatus _status;
 
         internal MessageDispatcherStatus Status => _status;
@@ -54,8 +54,12 @@ namespace Abc.Zebus.Dispatch
             foreach (var invokerLoader in _invokerLoaders)
             {
                 var loadedInvokers = invokerLoader.LoadMessageHandlerInvokers(typeSource);
+
                 foreach (var invoker in loadedInvokers)
                 {
+                    if (invoker.MessageType is null)
+                        continue;
+
                     if (_handlerFilter != null && !_handlerFilter(invoker.MessageHandlerType))
                         continue;
 

src/Abc.Zebus/Dispatch/MessageHandlerInvoker.cs

@@ -15,12 +15,12 @@ namespace Abc.Zebus.Dispatch
     {
         private readonly Instance _instance;
         private bool? _isSingleton;
-        private IBus _bus;
+        private IBus? _bus;
 
         [ThreadStatic]
-        private static MessageContextAwareBus _dispatchBus;
+        private static MessageContextAwareBus? _dispatchBus;
 
-        protected MessageHandlerInvoker(Type handlerType, Type messageType, bool? shouldBeSubscribedOnStartup = null)
+        protected MessageHandlerInvoker(Type handlerType, Type? messageType, bool? shouldBeSubscribedOnStartup = null)
         {
             MessageHandlerType = handlerType;
             DispatchQueueName = DispatchQueueNameScanner.GetQueueName(handlerType);
@@ -32,7 +32,7 @@ namespace Abc.Zebus.Dispatch
         }
 
         public Type MessageHandlerType { get; }
-        public Type MessageType { get; }
+        public Type? MessageType { get; }
         public MessageTypeId MessageTypeId { get; }
         public bool ShouldBeSubscribedOnStartup { get; }
         public string DispatchQueueName { get; }
@@ -60,8 +60,11 @@ namespace Abc.Zebus.Dispatch
             return MessageShouldBeSubscribedOnStartup(messageType, GetExplicitSubscriptionMode(handlerType));
         }
 
-        internal static bool MessageShouldBeSubscribedOnStartup(Type messageType, SubscriptionMode? subscriptionMode = null)
+        internal static bool MessageShouldBeSubscribedOnStartup(Type? messageType, SubscriptionMode? subscriptionMode = null)
         {
+            if (messageType is null)
+                return false;
+
             if (subscriptionMode != null)
                 return subscriptionMode == SubscriptionMode.Auto;
 
@@ -86,7 +89,7 @@ namespace Abc.Zebus.Dispatch
             if (IsHandlerSingleton(container))
                 return container.GetInstance(MessageHandlerType);
 
-            _bus = _bus ?? container.GetInstance<IBus>();
+            _bus ??= container.GetInstance<IBus>();
             if (_bus == null)
                 return container.GetInstance(MessageHandlerType);
 
@@ -114,8 +117,8 @@ namespace Abc.Zebus.Dispatch
         private static Instance CreateConstructorInstance(Type messageHandlerType)
         {
             var inst = new ConstructorInstance(messageHandlerType);
-            inst.Dependencies.Add<IBus>(new LambdaInstance<IBus>("Dispatch IBus", () => _dispatchBus));
-            inst.Dependencies.Add<MessageContext>(new LambdaInstance<MessageContext>("Dispatch MessageContext", () => _dispatchBus.MessageContext));
+            inst.Dependencies.Add<IBus>(new LambdaInstance<IBus>("Dispatch IBus", () => _dispatchBus!));
+            inst.Dependencies.Add<MessageContext>(new LambdaInstance<MessageContext>("Dispatch MessageContext", () => _dispatchBus!.MessageContext));
             return inst;
         }
 

src/Abc.Zebus/Dispatch/Pipes/AfterInvokeArgs.cs

@@ -5,11 +5,11 @@ namespace Abc.Zebus.Dispatch.Pipes
     public readonly struct AfterInvokeArgs
     {
         public readonly PipeInvocation Invocation;
-        public readonly object State;
+        public readonly object? State;
         public readonly bool IsFaulted;
-        public readonly Exception Exception;
+        public readonly Exception? Exception;
 
-        public AfterInvokeArgs(PipeInvocation invocation, object state, bool isFaulted, Exception exception)
+        public AfterInvokeArgs(PipeInvocation invocation, object? state, bool isFaulted, Exception? exception)
         {
             Invocation = invocation;
             State = state;

src/Abc.Zebus/Dispatch/Pipes/BeforeInvokeArgs.cs

@@ -17,7 +17,7 @@
 
         public readonly PipeInvocation Invocation;
 
-        public object State
+        public object? State
         {
             get => _stateRef.Value;
             set => _stateRef.Value = value;
@@ -25,7 +25,7 @@
 
         public class StateRef
         {
-            public object Value;
+            public object? Value;
         }
     }
 }

src/Abc.Zebus/Dispatch/Pipes/PipeAttribute.cs

@@ -11,6 +11,6 @@ namespace Abc.Zebus.Dispatch.Pipes
             PipeType = pipeType;
         }
 
-        public Type PipeType { get; private set; }
+        public Type PipeType { get; }
     }
-}
+}

src/Abc.Zebus/Dispatch/Pipes/PipeInvocation.cs

@@ -54,13 +54,13 @@ namespace Abc.Zebus.Dispatch.Pipes
             AfterInvoke(pipeStates, false, null);
         }
 
-        private object[] BeforeInvoke()
+        private object?[] BeforeInvoke()
         {
             if (_pipes.Count == 0)
-                return Array.Empty<object>();
+                return Array.Empty<object?>();
 
             var stateRef = new BeforeInvokeArgs.StateRef();
-            var pipeStates = new object[_pipes.Count];
+            var pipeStates = new object?[_pipes.Count];
             for (var pipeIndex = 0; pipeIndex < _pipes.Count; ++pipeIndex)
             {
                 var beforeInvokeArgs = new BeforeInvokeArgs(this, stateRef);
@@ -71,7 +71,7 @@ namespace Abc.Zebus.Dispatch.Pipes
             return pipeStates;
         }
 
-        private void AfterInvoke(object[] pipeStates, bool isFaulted, Exception exception)
+        private void AfterInvoke(object?[] pipeStates, bool isFaulted, Exception? exception)
         {
             for (var pipeIndex = _pipes.Count - 1; pipeIndex >= 0; --pipeIndex)
             {
@@ -88,7 +88,7 @@ namespace Abc.Zebus.Dispatch.Pipes
 
             if (runTask.Status == TaskStatus.Created)
             {
-                var exception = new InvalidProgramException($"{Invoker.MessageHandlerType.Name}.Handle({Invoker.MessageType.Name}) did not start the returned task");
+                var exception = new InvalidProgramException($"{Invoker.MessageHandlerType.Name}.Handle({Invoker.MessageType?.Name}) did not start the returned task");
                 runTask = Task.FromException(exception);
             }
 

src/Abc.Zebus/Dispatch/Pipes/PipeManager.cs

@@ -51,16 +51,16 @@ namespace Abc.Zebus.Dispatch.Pipes
             return new PipeInvocation(messageHandlerInvoker, messages, messageContext, pipes);
         }
 
-        public IEnumerable<IPipe> GetEnabledPipes(Type messageHandlerType) 
+        public IEnumerable<IPipe> GetEnabledPipes(Type messageHandlerType)
             => GetPipeList(messageHandlerType).EnabledPipes;
 
-        private PipeList GetPipeList(Type messageHandlerType) 
+        private PipeList GetPipeList(Type messageHandlerType)
             => _pipesByMessageType.GetOrAdd(messageHandlerType, _createPipeList);
 
-        private PipeList CreatePipeList(Type handlerType) 
+        private PipeList CreatePipeList(Type handlerType)
             => new PipeList(this, _pipeSources.SelectMany(x => x.GetPipes(handlerType)));
 
-        private bool IsPipeEnabled(IPipe pipe) 
+        private bool IsPipeEnabled(IPipe pipe)
             => !_disabledPipeNames.Contains(pipe.Name)
                && (pipe.IsAutoEnabled || _enabledPipeNames.Contains(pipe.Name));
 
@@ -77,7 +77,7 @@ namespace Abc.Zebus.Dispatch.Pipes
                 ReloadEnabledPipes();
             }
 
-            public IList<IPipe> EnabledPipes { get; private set; }
+            public IList<IPipe> EnabledPipes { get; private set; } = default!;
 
             internal void ReloadEnabledPipes()
             {

src/Abc.Zebus/DomainException.cs

@@ -13,7 +13,7 @@ namespace Abc.Zebus
     {
         public int ErrorCode { get; private set; }
 
-        public DomainException(Exception innerException, int errorCode, string message)
+        public DomainException(Exception? innerException, int errorCode, string message)
             : base(message, innerException)
         {
             ErrorCode = errorCode;
@@ -24,7 +24,7 @@ namespace Abc.Zebus
         {
         }
 
-        public DomainException(string message, Exception innerException)
+        public DomainException(string message, Exception? innerException)
             : this(innerException, ErrorStatus.UnknownError.Code, message)
         {
         }

src/Abc.Zebus/ErrorStatus.cs

@@ -1,5 +1,3 @@
-using JetBrains.Annotations;
-
 namespace Abc.Zebus
 {
     internal class ErrorStatus
@@ -8,11 +6,9 @@ namespace Abc.Zebus
         public static readonly ErrorStatus UnknownError = new ErrorStatus(1, null);
 
         public int Code { get; }
+        public string? Message { get; }
 
-        [CanBeNull]
-        public string Message { get; }
-
-        public ErrorStatus(int code, string message)
+        public ErrorStatus(int code, string? message)
         {
             Code = code;
             Message = message;

src/Abc.Zebus/Hosting/PeriodicActionHostInitializer.cs

@@ -12,13 +12,13 @@ namespace Abc.Zebus.Hosting
     {
         protected readonly ILog _logger;
         private readonly IBus _bus;
-        private readonly Func<DateTime> _dueTimeUtcFunc;
-        private Timer _timer;
+        private readonly Func<DateTime>? _dueTimeUtcFunc;
+        private Timer? _timer;
         private int _exceptionCount;
         private DateTime _nextInvocationUtc;
         private DateTime _pauseEndTimeUtc;
 
-        protected PeriodicActionHostInitializer(IBus bus, TimeSpan period, Func<DateTime> dueTimeUtcFunc = null)
+        protected PeriodicActionHostInitializer(IBus bus, TimeSpan period, Func<DateTime>? dueTimeUtcFunc = null)
         {
             _logger = LogManager.GetLogger(GetType());
             _bus = bus;

src/Abc.Zebus/IBus.cs

@@ -12,7 +12,7 @@ namespace Abc.Zebus
         void Configure(PeerId peerId, string environment);
 
         void Publish(IEvent message);
-        
+
         Task<CommandResult> Send(ICommand message);
         Task<CommandResult> Send(ICommand message, Peer peer);
 
@@ -20,9 +20,9 @@ namespace Abc.Zebus
         Task<IDisposable> SubscribeAsync(SubscriptionRequest request, Action<IMessage> handler);
 
         void Reply(int errorCode);
-        void Reply(int errorCode, string message);
-        void Reply(IMessage response);
-        
+        void Reply(int errorCode, string? message);
+        void Reply(IMessage? response);
+
         void Start();
         void Stop();
 

src/Abc.Zebus/Lotus/CustomProcessingFailed.cs

@@ -21,7 +21,7 @@ namespace Abc.Zebus.Lotus
         public PeerId? FailingPeerIdOverride { get; set; }
 
         [ProtoMember(5, IsRequired = false)]
-        public string DetailsJson { get; set; }
+        public string? DetailsJson { get; set; }
 
         public CustomProcessingFailed(string sourceTypeFullName, string exceptionMessage)
             : this(sourceTypeFullName, exceptionMessage, SystemDateTime.UtcNow)
@@ -35,7 +35,7 @@ namespace Abc.Zebus.Lotus
             ExceptionUtcTime = exceptionUtcTime;
         }
 
-        public CustomProcessingFailed WithDetails(object details)
+        public CustomProcessingFailed WithDetails(object? details)
         {
             DetailsJson = details != null ? JsonConvert.SerializeObject(details) : null;
             return this;

src/Abc.Zebus/Lotus/MessageProcessingFailed.cs

@@ -22,7 +22,7 @@ namespace Abc.Zebus.Lotus
         [ProtoMember(5, IsRequired = false)]
         public readonly string[] FailingHandlerNames;
 
-        public MessageProcessingFailed(TransportMessage failingMessage, string failingMessageJson, string exceptionMessage, DateTime exceptionUtcTime, string[] failingHandlerNames)
+        public MessageProcessingFailed(TransportMessage failingMessage, string failingMessageJson, string exceptionMessage, DateTime exceptionUtcTime, string[]? failingHandlerNames)
         {
             FailingMessage = failingMessage;
             FailingMessageJson = failingMessageJson;

src/Abc.Zebus/Lotus/ReplayMessageHandler.cs

@@ -1,4 +1,5 @@
-using Abc.Zebus.Dispatch;
+using System;
+using Abc.Zebus.Dispatch;
 
 namespace Abc.Zebus.Lotus
 {
@@ -15,8 +16,10 @@ namespace Abc.Zebus.Lotus
 
         public void Handle(ReplayMessageCommand message)
         {
-            var dispatch = _dispatchFactory.CreateMessageDispatch(message.MessageToReplay);
+            var dispatch = _dispatchFactory.CreateMessageDispatch(message.MessageToReplay)
+                           ?? throw new InvalidOperationException($"Could not dispatch message of type {message.MessageToReplay.MessageTypeId.FullName}");
+
             _dispatcher.Dispatch(dispatch, message.ShouldApplyToHandler);
         }
     }
-}
+}

src/Abc.Zebus/MessageContext.cs

@@ -11,34 +11,34 @@ namespace Abc.Zebus
         internal static readonly string CurrentUserName = Environment.UserName;
 
         [ThreadStatic]
-        private static MessageContext _current;
+        private static MessageContext? _current;
 
         [ThreadStatic]
-        private static string _outboundInitiatorOverride;
+        private static string? _outboundInitiatorOverride;
 
         public virtual int ReplyCode { get; internal set; }
-        public virtual string ReplyMessage { get; internal set; }
-        public virtual IMessage ReplyResponse { get; internal set; }
+        public virtual string? ReplyMessage { get; internal set; }
+        public virtual IMessage? ReplyResponse { get; internal set; }
         public virtual MessageId MessageId { get; private set; }
-        public virtual OriginatorInfo Originator { get; private set; }
+        public virtual OriginatorInfo Originator { get; private set; } = default!;
 
         public PeerId SenderId => Originator.SenderId;
         public string SenderEndPoint => Originator.SenderEndPoint;
-        public string InitiatorUserName => Originator.InitiatorUserName;
+        public string? InitiatorUserName => Originator.InitiatorUserName;
 
         public Peer GetSender()
         {
             return new Peer(SenderId, SenderEndPoint);
         }
 
-        public static MessageContext Current => _current;
+        public static MessageContext? Current => _current;
 
         public static IDisposable SetCurrent(MessageContext context)
             => new MessageContextScope(context);
 
         public static IDisposable OverrideInitiatorUsername(string username)
         {
-            string currentInitiatorUsername = null,
+            string? currentInitiatorUsername = null,
                 previousOverride = _outboundInitiatorOverride;
 
             var current = Current;
@@ -95,7 +95,7 @@ namespace Abc.Zebus
             };
         }
 
-        internal static string GetInitiatorUserName()
+        internal static string? GetInitiatorUserName()
         {
             var current = Current;
             return current != null ? current.Originator.InitiatorUserName : (_outboundInitiatorOverride ?? CurrentUserName);
@@ -111,7 +111,7 @@ namespace Abc.Zebus
 
         private class MessageContextScope : IDisposable
         {
-            private readonly MessageContext _previous;
+            private readonly MessageContext? _previous;
             private int _disposed;
 
             public MessageContextScope(MessageContext context)

src/Abc.Zebus/MessageId.cs

@@ -26,7 +26,7 @@ namespace Abc.Zebus
 
         public bool Equals(MessageId other) => Value.Equals(other.Value);
 
-        public override bool Equals(object obj) => obj is MessageId id && Equals(id);
+        public override bool Equals(object? obj) => obj is MessageId id && Equals(id);
 
         public override int GetHashCode() => Value.GetHashCode();
 

src/Abc.Zebus/MessageTypeDescriptor.cs

@@ -5,9 +5,9 @@ namespace Abc.Zebus
 {
     internal class MessageTypeDescriptor
     {
-        public static readonly MessageTypeDescriptor Null = new MessageTypeDescriptor(null, null, false, false);
+        public static readonly MessageTypeDescriptor Null = new MessageTypeDescriptor(null!, null, false, false); // TODO replace with null
 
-        private MessageTypeDescriptor(string fullName, Type messageType, bool isPersistent, bool isInfrastructure)
+        private MessageTypeDescriptor(string fullName, Type? messageType, bool isPersistent, bool isInfrastructure)
         {
             FullName = fullName;
             MessageType = messageType;
@@ -16,7 +16,7 @@ namespace Abc.Zebus
         }
 
         public string FullName { get; }
-        public Type MessageType { get; }
+        public Type? MessageType { get; }
         public bool IsPersistent { get; }
         public bool IsInfrastructure { get; }
 
@@ -29,4 +29,4 @@ namespace Abc.Zebus
             return new MessageTypeDescriptor(fullName, messageType, isPersistent, isInfrastructure);
         }
     }
-}
+}

src/Abc.Zebus/MessageTypeId.cs

@@ -1,5 +1,4 @@
 using System;
-using System.Diagnostics.CodeAnalysis;
 using JetBrains.Annotations;
 using ProtoBuf;
 
@@ -13,22 +12,22 @@ namespace Abc.Zebus
         public static readonly MessageTypeId PersistenceStopping = new MessageTypeId("Abc.Zebus.PersistentTransport.PersistenceStopping");
         public static readonly MessageTypeId PersistenceStoppingAck = new MessageTypeId("Abc.Zebus.PersistentTransport.PersistenceStoppingAck");
 
-        private readonly MessageTypeDescriptor _descriptor;
+        private readonly MessageTypeDescriptor? _descriptor;
 
-        public MessageTypeId(Type messageType)
+        public MessageTypeId(Type? messageType)
         {
             _descriptor = MessageUtil.GetMessageTypeDescriptor(messageType);
         }
 
-        public MessageTypeId(string fullName)
+        public MessageTypeId(string? fullName)
         {
             _descriptor = MessageUtil.GetMessageTypeDescriptor(fullName);
         }
 
-        public string FullName => _descriptor?.FullName;
+        public string? FullName => _descriptor?.FullName;
 
         [System.Diagnostics.Contracts.Pure]
-        public Type GetMessageType() => _descriptor?.MessageType;
+        public Type? GetMessageType() => _descriptor?.MessageType;
 
         [System.Diagnostics.Contracts.Pure]
         public bool IsInfrastructure() => _descriptor?.IsInfrastructure ?? false;
@@ -38,14 +37,16 @@ namespace Abc.Zebus
 
         public override string ToString()
         {
+            if (FullName is null)
+                return "(unknown type)";
+
             var lastDotIndex = FullName.LastIndexOf('.');
             return lastDotIndex != -1 ? FullName.Substring(lastDotIndex + 1) : FullName;
         }
 
         public bool Equals(MessageTypeId other) => _descriptor == other._descriptor;
-        public override bool Equals(object obj) => obj is MessageTypeId messageTypeId && Equals(messageTypeId);
+        public override bool Equals(object? obj) => obj is MessageTypeId messageTypeId && Equals(messageTypeId);
 
-        [SuppressMessage("ReSharper", "NonReadonlyMemberInGetHashCode")]
         public override int GetHashCode() => _descriptor?.GetHashCode() ?? 0;
 
         public static bool operator ==(MessageTypeId left, MessageTypeId right) => left.Equals(right);
@@ -56,7 +57,7 @@ namespace Abc.Zebus
         internal struct ProtobufSurrogate
         {
             [ProtoMember(1, IsRequired = true)]
-            private string FullName { get; set; }
+            private string? FullName { get; set; }
 
             private ProtobufSurrogate(MessageTypeId typeId)
             {

src/Abc.Zebus/MessageUtil.cs

@@ -27,7 +27,7 @@ namespace Abc.Zebus
             return new MessageTypeId(messageType);
         }
 
-        internal static MessageTypeDescriptor GetMessageTypeDescriptor(string fullName)
+        internal static MessageTypeDescriptor GetMessageTypeDescriptor(string? fullName)
         {
             if (fullName == null)
                 return MessageTypeDescriptor.Null;
@@ -40,7 +40,7 @@ namespace Abc.Zebus
             return MessageTypeDescriptor.Load(fullName);
         }
 
-        internal static MessageTypeDescriptor GetMessageTypeDescriptor(Type messageType)
+        internal static MessageTypeDescriptor GetMessageTypeDescriptor(Type? messageType)
         {
             if (messageType == null)
                 return MessageTypeDescriptor.Null;

src/Abc.Zebus/Peer.cs

@@ -38,14 +38,12 @@ namespace Abc.Zebus
         [UsedImplicitly]
         private Peer()
         {
+            EndPoint = default!;
         }
 
         public override string ToString() => $"{Id}, {EndPoint}";
 
         public string GetMachineNameFromEndPoint()
-        {
-            var uri = new Uri(EndPoint);
-            return uri.Host;
-        }
+            => new Uri(EndPoint).Host;
     }
-}
+}

src/Abc.Zebus/PeerId.cs

@@ -18,7 +18,7 @@ namespace Abc.Zebus
         }
 
         public bool Equals(PeerId other) => string.Equals(_value, other._value);
-        public override bool Equals(object obj) => obj is PeerId && Equals((PeerId)obj);
+        public override bool Equals(object? obj) => obj is PeerId peerId && Equals(peerId);
 
         public override int GetHashCode() => _value?.GetHashCode() ?? 0;
 

src/Abc.Zebus/Persistence/PersistentTransport.cs

@@ -28,10 +28,10 @@ namespace Abc.Zebus.Persistence
         private readonly IPeerDirectory _peerDirectory;
         private readonly IMessageSendingStrategy _messageSendingStrategy;
         private readonly bool _isPersistent;
-        private BlockingCollection<TransportMessage> _pendingReceives;
+        private BlockingCollection<TransportMessage> _pendingReceives = new BlockingCollection<TransportMessage>();
         private bool _isRunning;
-        private Phase _phase;
-        private Thread _receptionThread;
+        private Phase _phase = default!;
+        private Thread? _receptionThread;
         private Guid? _currentReplayId;
         private volatile bool _persistenceIsDown;
 
@@ -48,7 +48,7 @@ namespace Abc.Zebus.Persistence
             _innerTransport.MessageReceived += OnTransportMessageReceived;
         }
 
-        public event Action<TransportMessage> MessageReceived;
+        public event Action<TransportMessage>? MessageReceived;
 
         public PeerId PeerId => _innerTransport.PeerId;
 
@@ -92,8 +92,7 @@ namespace Abc.Zebus.Persistence
 
             _logger.InfoFormat("Sending {0} enqueued messages to the persistence", _messagesWaitingForPersistence.Count);
 
-            TransportMessage messageToSend;
-            while (_messagesWaitingForPersistence.TryTake(out messageToSend))
+            while (_messagesWaitingForPersistence.TryTake(out var messageToSend))
             {
                 _innerTransport.Send(messageToSend, persistencePeers, new SendContext());
             }
@@ -132,7 +131,8 @@ namespace Abc.Zebus.Persistence
 
         public void Start()
         {
-            _pendingReceives = new BlockingCollection<TransportMessage>();
+            if (_pendingReceives.IsAddingCompleted)
+                _pendingReceives = new BlockingCollection<TransportMessage>();
 
             _phase.OnStart();
 
@@ -242,7 +242,7 @@ namespace Abc.Zebus.Persistence
         {
             if (_replayMessageTypeIds.Contains(transportMessage.MessageTypeId))
             {
-                var replayEvent = (IReplayEvent)_serializer.ToMessage(transportMessage);
+                var replayEvent = (IReplayEvent)_serializer.ToMessage(transportMessage)!;
                 if (replayEvent.ReplayId == _currentReplayId)
                     _phase.OnReplayEvent(replayEvent);
 

src/Abc.Zebus/Routing/BindingKey.cs

@@ -3,8 +3,8 @@ using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Linq;
 using System.Reflection;
-using JetBrains.Annotations;
 using Abc.Zebus.Util.Extensions;
+using JetBrains.Annotations;
 using ProtoBuf;
 
 namespace Abc.Zebus.Routing
@@ -16,11 +16,11 @@ namespace Abc.Zebus.Routing
         private const string _sharp = "#";
         public static readonly BindingKey Empty = new BindingKey();
 
-        private static readonly ConcurrentDictionary<Type, BindingKeyBuilder> _builders = new ConcurrentDictionary<Type, BindingKeyBuilder>();
-        private static readonly Func<Type, BindingKeyBuilder> _bindingKeyBuilderFactory = CreateBuilder;
+        private static readonly ConcurrentDictionary<Type, BindingKeyBuilder?> _builders = new ConcurrentDictionary<Type, BindingKeyBuilder?>();
+        private static readonly Func<Type, BindingKeyBuilder?> _bindingKeyBuilderFactory = CreateBuilder;
 
         [ProtoMember(1, IsRequired = true)]
-        private readonly string[] _parts;
+        private readonly string[]? _parts;
 
         public BindingKey(params string[] parts)
         {
@@ -35,20 +35,20 @@ namespace Abc.Zebus.Routing
         public bool IsEmpty => _parts == null || _parts.Length == 1 && IsSharp(0);
 
         public bool IsSharp(int index)
-            => _parts[index] == _sharp;
+            => _parts?[index] == _sharp;
 
         public bool IsStar(int index)
-            => _parts[index] == _star;
+            => _parts?[index] == _star;
 
         [Pure]
-        public string GetPart(int index)
-            => index < PartCount ? _parts[index] : null;
+        public string? GetPart(int index)
+            => index < PartCount ? _parts![index] : null;
 
         [Pure]
         public IEnumerable<string> GetParts()
             => _parts?.ToList() ?? Enumerable.Empty<string>();
 
-        public override bool Equals(object obj)
+        public override bool Equals(object? obj)
             => obj is BindingKey other && Equals(other);
 
         public bool Equals(BindingKey other)
@@ -96,10 +96,10 @@ namespace Abc.Zebus.Routing
         internal static BindingKey Create(Type messageType, IDictionary<string, string> fieldValues)
             => GetBindingKeyBuilder(messageType)?.BuildKey(fieldValues) ?? Empty;
 
-        private static BindingKeyBuilder GetBindingKeyBuilder(Type messageType)
+        private static BindingKeyBuilder? GetBindingKeyBuilder(Type messageType)
             => _builders.GetOrAdd(messageType, _bindingKeyBuilderFactory);
 
-        private static BindingKeyBuilder CreateBuilder(Type messageType)
+        private static BindingKeyBuilder? CreateBuilder(Type messageType)
         {
             if (!Attribute.IsDefined(messageType, typeof(Routable)))
                 return null;

src/Abc.Zebus/Routing/BindingKeyPredicateBuilder.cs

@@ -10,16 +10,10 @@ namespace Abc.Zebus.Routing
 {
     public class BindingKeyPredicateBuilder : IBindingKeyPredicateBuilder
     {
-        private static readonly MethodInfo _toStringMethod = typeof(object).GetMethod("ToString");
-        private static readonly MethodInfo _toStringWithFormatMethod = typeof(IConvertible).GetMethod("ToString");
+        private static readonly MethodInfo _toStringMethod = typeof(object).GetMethod(nameof(ToString));
+        private static readonly MethodInfo _toStringWithFormatMethod = typeof(IConvertible).GetMethod(nameof(IConvertible.ToString));
         private readonly ConcurrentDictionary<Type, CacheItem> _cacheItems = new ConcurrentDictionary<Type, CacheItem>();
 
-        class CacheItem
-        {
-            public ParameterExpression ParameterExpression { get; set; }
-            public IList<MethodCallExpression> MembersToStringExpressions { get; set; }
-        }
-
         public Func<IMessage, bool> GetPredicate(Type messageType, BindingKey bindingKey)
         {
             if (bindingKey.IsEmpty)
@@ -44,8 +38,8 @@ namespace Abc.Zebus.Routing
 
             if (!subPredicates.Any())
                 return _ => true;
-            
-            var finalExpression = subPredicates.Aggregate((Expression)null, (final, exp) => final == null ? exp : Expression.AndAlso(final, exp));
+
+            var finalExpression = subPredicates.Aggregate((Expression?)null, (final, exp) => final == null ? exp : Expression.AndAlso(final, exp))!;
             return (Func<IMessage, bool>)Expression.Lambda(finalExpression, cacheItem.ParameterExpression).Compile();
         }
 
@@ -54,7 +48,7 @@ namespace Abc.Zebus.Routing
             return _cacheItems.GetOrAdd(messageType, type =>
             {
                 var routingMembers = type.GetMembers(BindingFlags.Public | BindingFlags.Instance)
-                                         .Select(x => new MemberExtendedInfo { Member = x, Attribute = x.GetCustomAttribute<RoutingPositionAttribute>(true) })
+                                         .Select(x => new MemberExtendedInfo(x))
                                          .Where(x => x.Attribute != null)
                                          .OrderBy(x => x.Attribute.Position)
                                          .ToList();
@@ -62,11 +56,7 @@ namespace Abc.Zebus.Routing
                 var parameterExpression = Expression.Parameter(typeof(IMessage), "m");
                 var castedMessage = Expression.Convert(parameterExpression, messageType);
 
-                return new CacheItem
-                {
-                    ParameterExpression = parameterExpression,
-                    MembersToStringExpressions = routingMembers.Select(x => GenerateMemberToStringExpression(castedMessage, x)).ToList()
-                };
+                return new CacheItem(parameterExpression, routingMembers.Select(x => GenerateMemberToStringExpression(castedMessage, x)));
             });
         }
 
@@ -95,11 +85,29 @@ namespace Abc.Zebus.Routing
                 : Expression.Call(memberAccessor(parameterExpression), _toStringMethod);
             return getMemberValue;
         }
-    }
 
-    public class MemberExtendedInfo
-    {
-        public MemberInfo Member { get; set; }
-        public RoutingPositionAttribute Attribute { get; set; }
+        private class CacheItem
+        {
+            public ParameterExpression ParameterExpression { get; }
+            public IList<MethodCallExpression> MembersToStringExpressions { get; }
+
+            public CacheItem(ParameterExpression parameterExpression, IEnumerable<MethodCallExpression> membersToStringExpressions)
+            {
+                ParameterExpression = parameterExpression;
+                MembersToStringExpressions = membersToStringExpressions.ToList();
+            }
+        }
+
+        private class MemberExtendedInfo
+        {
+            public MemberInfo Member { get; }
+            public RoutingPositionAttribute Attribute { get; }
+
+            public MemberExtendedInfo(MemberInfo member)
+            {
+                Member = member;
+                Attribute = member.GetCustomAttribute<RoutingPositionAttribute>(true);
+            }
+        }
     }
-}
+}

src/Abc.Zebus/Routing/BindingKeyToken.cs

@@ -7,8 +7,8 @@ namespace Abc.Zebus.Routing
 {
     internal class BindingKeyToken
     {
-        private static readonly MethodInfo _toStringMethod = typeof(object).GetMethod("ToString");
-        private static readonly MethodInfo _toStringWithFormatMethod = typeof(IConvertible).GetMethod("ToString");
+        private static readonly MethodInfo _toStringMethod = typeof(object).GetMethod(nameof(ToString));
+        private static readonly MethodInfo _toStringWithFormatMethod = typeof(IConvertible).GetMethod(nameof(IConvertible.ToString));
         private readonly Func<IMessage, string> _valueAccessorFunc;
 
         public BindingKeyToken(int position, Type messageType, FieldInfo fieldInfo)
@@ -57,4 +57,4 @@ namespace Abc.Zebus.Routing
             return (Func<IMessage, string>)lambda.Compile();
         }
     }
-}
+}

src/Abc.Zebus/Scan/DispatchQueueNameScanner.cs

@@ -34,7 +34,7 @@ namespace Abc.Zebus.Scan
             var assemblyCache = _assemblyCaches.GetOrAdd(type.Assembly, x => new AssemblyCache(x));
             return assemblyCache.GetQueueNameFromNamespace(type.Namespace);
         }
- 
+
         private class AssemblyCache
         {
             private readonly Dictionary<string, string> _knownQueueNames = new Dictionary<string, string>();
@@ -62,7 +62,7 @@ namespace Abc.Zebus.Scan
                 var queueName = _knownQueueNames.GetValueOrDefault(@namespace);
                 if (queueName != null)
                     return queueName;
-                 
+
                 var parentNamespace = @namespace.Qualifier();
                 if (parentNamespace.Length == @namespace.Length)
                     return DefaultQueueName;
@@ -71,4 +71,4 @@ namespace Abc.Zebus.Scan
             }
         }
     }
-}
+}

src/Abc.Zebus/Scan/MessageHandlerInvokerLoader.cs

@@ -20,7 +20,7 @@ namespace Abc.Zebus.Scan
             Container = container;
         }
 
-        protected IContainer Container { get; private set; }
+        protected IContainer Container { get; }
 
         public IEnumerable<IMessageHandlerInvoker> LoadMessageHandlerInvokers(TypeSource typeSource)
         {

src/Abc.Zebus/Serialization/IMessageSerializer.cs

@@ -1,11 +1,12 @@
-using System.IO;
+using System.Diagnostics.CodeAnalysis;
+using System.IO;
 
 namespace Abc.Zebus.Serialization
 {
     public interface IMessageSerializer
     {
         Stream Serialize(IMessage message);
-        IMessage Deserialize(MessageTypeId messageTypeId, Stream stream);
-        bool TryClone(IMessage message, out IMessage clone);
+        IMessage? Deserialize(MessageTypeId messageTypeId, Stream stream);
+        bool TryClone(IMessage message, [MaybeNullWhen(false)] out IMessage clone);
     }
 }

src/Abc.Zebus/Serialization/MessageSerializer.cs

@@ -7,7 +7,7 @@ namespace Abc.Zebus.Serialization
     {
         private static readonly ILog _log = LogManager.GetLogger(typeof(MessageSerializer));
 
-        public IMessage Deserialize(MessageTypeId messageTypeId, Stream stream)
+        public IMessage? Deserialize(MessageTypeId messageTypeId, Stream stream)
         {
             var messageType = messageTypeId.GetMessageType();
             if (messageType != null)
@@ -15,7 +15,6 @@ namespace Abc.Zebus.Serialization
 
             _log.WarnFormat("Could not find message type: {0}", messageTypeId.FullName);
             return null;
-
         }
 
         public Stream Serialize(IMessage message)
@@ -28,6 +27,6 @@ namespace Abc.Zebus.Serialization
         }
 
         public bool TryClone(IMessage message, out IMessage clone)
-            => Serializer.TryClone(message, out  clone);
+            => Serializer.TryClone(message, out clone!);
     }
 }

src/Abc.Zebus/Serialization/MessageSerializerExtensions.cs

@@ -8,19 +8,21 @@ namespace Abc.Zebus.Serialization
     {
         public static TransportMessage ToTransportMessage(this IMessageSerializer serializer, IMessage message, PeerId peerId, string peerEndPoint)
         {
-            var persistMessageCommand = message as PersistMessageCommand;
-            if (persistMessageCommand != null)
+            if (message is PersistMessageCommand persistMessageCommand)
                 return ToTransportMessage(persistMessageCommand);
 
             return new TransportMessage(message.TypeId(), serializer.Serialize(message), peerId, peerEndPoint);
         }
 
-        public static IMessage ToMessage(this IMessageSerializer serializer, TransportMessage transportMessage)
+        public static IMessage? ToMessage(this IMessageSerializer serializer, TransportMessage transportMessage)
         {
+            if (transportMessage.Content is null)
+                return null;
+
             return ToMessage(serializer, transportMessage, transportMessage.MessageTypeId, transportMessage.Content);
         }
 
-        public static IMessage ToMessage(this IMessageSerializer serializer, TransportMessage transportMessage, MessageTypeId messageTypeId, Stream content)
+        public static IMessage? ToMessage(this IMessageSerializer serializer, TransportMessage transportMessage, MessageTypeId messageTypeId, Stream content)
         {
             if (transportMessage.IsPersistTransportMessage)
                 return ToPersistMessageCommand(transportMessage);
@@ -32,6 +34,6 @@ namespace Abc.Zebus.Serialization
             => persistMessageCommand.TransportMessage.ToPersistTransportMessage(persistMessageCommand.Targets);
 
         private static IMessage ToPersistMessageCommand(TransportMessage transportMessage)
-            => new PersistMessageCommand(transportMessage.UnpackPersistTransportMessage(), transportMessage.PersistentPeerIds);
+            => new PersistMessageCommand(transportMessage.UnpackPersistTransportMessage(), transportMessage.PersistentPeerIds!);
     }
 }

src/Abc.Zebus/Serialization/PeerIdConverter.cs

@@ -7,18 +7,18 @@ namespace Abc.Zebus.Serialization
     [UsedImplicitly]
     public class PeerIdConverter : JsonConverter
     {
-        public override void WriteJson(JsonWriter writer, object value, JsonSerializer serializer)
+        public override void WriteJson(JsonWriter writer, object? value, JsonSerializer serializer)
         {
-            var peerId = (PeerId)value;
+            var peerId = (PeerId)value!;
             writer.WriteValue(peerId.ToString());
         }
 
-        public override object ReadJson(JsonReader reader, Type objectType, object existingValue, JsonSerializer serializer)
+        public override object ReadJson(JsonReader reader, Type objectType, object? existingValue, JsonSerializer serializer)
         {
             if (reader.TokenType != JsonToken.String)
                 return Activator.CreateInstance(objectType); // objectType can be Nullable<PeerId>
 
-            var value = reader.Value.ToString();
+            var value = reader.Value!.ToString();
             return new PeerId(value);
         }
 

src/Abc.Zebus/Serialization/Protobuf/InvalidProtocolBufferException.cs

@@ -46,7 +46,7 @@ namespace Abc.Zebus.Serialization.Protobuf
         {
         }
 
-        internal InvalidProtocolBufferException(string message, Exception innerException)
+        internal InvalidProtocolBufferException(string message, Exception? innerException)
             : base(message, innerException)
         {
         }
@@ -126,4 +126,4 @@ namespace Abc.Zebus.Serialization.Protobuf
                 "Stream of protocol messages had invalid tag. Expected tag is length-delimited field 1.");
         }
     }
-}
+}

src/Abc.Zebus/Serialization/ProtocolBufferSerializationException.cs

@@ -4,12 +4,12 @@ namespace Abc.Zebus.Serialization
 {
     public class ProtocolBufferSerializationException : Exception
     {
-        public object MessageToSerialize { get; private set; }
+        public object MessageToSerialize { get; }
 
-        public ProtocolBufferSerializationException(object message, Exception exception) 
+        public ProtocolBufferSerializationException(object message, Exception? exception)
             : base("Unable to serialize message. See inner exception for more details", exception)
         {
             MessageToSerialize = message;
         }
     }
-}
+}

src/Abc.Zebus/Serialization/Serializer.cs

@@ -31,31 +31,32 @@ namespace Abc.Zebus.Serialization
             }
         }
 
-        public static object Deserialize(Type messageType, Stream stream)
+        [return: NotNullIfNotNull("messageType")]
+        public static object? Deserialize(Type? messageType, Stream stream)
         {
-            if (messageType == null)
+            if (messageType is null)
                 return null;
 
             stream.Position = 0; // Reset position
 
-            object obj = null;
+            object? obj = null;
             if (!HasParameterLessConstructor(messageType) && messageType != typeof(string))
                 obj = FormatterServices.GetUninitializedObject(messageType);
 
             return RuntimeTypeModel.Default.Deserialize(stream, obj, messageType);
         }
 
-        public static bool TryClone<T>(T message, out T clone)
+        public static bool TryClone<T>(T? message, [NotNullWhen(true)] out T? clone)
             where T : class
         {
             var messageType = message?.GetType();
-            if (messageType!= null && RuntimeTypeModel.Default.CanSerialize(messageType))
+            if (messageType != null && RuntimeTypeModel.Default.CanSerialize(messageType))
             {
                 // Cannot use the DeepClone method as it doesn't handle classes without a parameterless constructor
 
                 using (var ms = new MemoryStream())
                 {
-                    Serialize(ms, message);
+                    Serialize(ms, message!);
                     ms.Position = 0;
                     clone = (T)Deserialize(messageType, ms);
                 }

src/Abc.Zebus/Subscription.cs

@@ -1,9 +1,9 @@
 using System;
 using System.Collections.Generic;
 using System.ComponentModel;
+using System.Diagnostics.CodeAnalysis;
 using System.Linq.Expressions;
 using System.Reflection;
-using System.Text.RegularExpressions;
 using Abc.Zebus.Directory;
 using Abc.Zebus.Routing;
 using Abc.Zebus.Util;
@@ -15,7 +15,7 @@ namespace Abc.Zebus
     [ProtoContract]
     public class Subscription : IEquatable<Subscription>
     {
-        private static readonly MethodInfo _wildCardTokenMethod = typeof(Builder).GetMethod("Any");
+        private static readonly MethodInfo _wildCardTokenMethod = typeof(Builder).GetMethod(nameof(Builder.Any));
         private int _computedHashCode;
 
         [ProtoMember(1, IsRequired = true)]
@@ -65,7 +65,7 @@ namespace Abc.Zebus
             return routingKey.PartCount == BindingKey.PartCount;
         }
 
-        public bool Equals(Subscription other)
+        public bool Equals(Subscription? other)
         {
             if (other == null)
                 return false;
@@ -73,18 +73,17 @@ namespace Abc.Zebus
             return MessageTypeId.Equals(other.MessageTypeId) && BindingKey.Equals(other.BindingKey);
         }
 
-        public override bool Equals(object obj) => Equals(obj as Subscription);
+        public override bool Equals(object? obj) => Equals(obj as Subscription);
 
+        [SuppressMessage("ReSharper", "NonReadonlyMemberInGetHashCode")]
         public override int GetHashCode()
         {
             unchecked
             {
-                // ReSharper disable NonReadonlyMemberInGetHashCode
                 if (_computedHashCode == 0)
                     _computedHashCode = (MessageTypeId.GetHashCode() * 397) ^ BindingKey.GetHashCode();
 
                 return _computedHashCode;
-                // ReSharper restore NonReadonlyMemberInGetHashCode
             }
         }
 
@@ -122,7 +121,7 @@ namespace Abc.Zebus
             return new Subscription(MessageUtil.TypeId<TMessage>(), BindingKey.Create(typeof(TMessage), parameterValues));
         }
 
-        private static object GetExpressionValue(Expression expression)
+        private static object? GetExpressionValue(Expression expression)
         {
             if (expression.NodeType != ExpressionType.Call)
                 return Expression.Lambda(expression).Compile().DynamicInvoke();
@@ -219,10 +218,9 @@ namespace Abc.Zebus
 
         private static void AddFieldValueFromBinaryExpression<TMessage>(Dictionary<string, string> fieldValues, BinaryExpression binaryExpression)
         {
-            MemberExpression memberExpression;
             Expression memberValueExpression;
 
-            if (TryGetMessageMemberExpression<TMessage>(binaryExpression.Right, out memberExpression))
+            if (TryGetMessageMemberExpression<TMessage>(binaryExpression.Right, out var memberExpression))
             {
                 memberValueExpression = binaryExpression.Left;
             }
@@ -244,7 +242,7 @@ namespace Abc.Zebus
             fieldValues.Add(memberName, valueAsText);
         }
 
-        private static bool TryGetMessageMemberExpression<TMessage>(Expression expression, out MemberExpression memberExpression)
+        private static bool TryGetMessageMemberExpression<TMessage>(Expression? expression, [NotNullWhen(true)] out MemberExpression? memberExpression)
         {
             memberExpression = expression as MemberExpression;
             if (memberExpression != null)
@@ -286,7 +284,7 @@ namespace Abc.Zebus
         [EditorBrowsable(EditorBrowsableState.Never), UsedImplicitly]
         public class Builder
         {
-            public T Any<T>() => default(T);
+            public T Any<T>() => default!;
         }
     }
 }

src/Abc.Zebus/SubscriptionRequest.cs

@@ -1,6 +1,5 @@
 using System;
 using System.Collections.Generic;
-using JetBrains.Annotations;
 using Abc.Zebus.Util.Extensions;
 
 namespace Abc.Zebus
@@ -13,8 +12,7 @@ namespace Abc.Zebus
 
         public bool ThereIsNoHandlerButIKnowWhatIAmDoing { get; set; }
 
-        [CanBeNull]
-        internal SubscriptionRequestBatch Batch { get; private set; }
+        internal SubscriptionRequestBatch? Batch { get; private set; }
 
         internal bool IsSubmitted { get; private set; }
         internal int? SubmissionSubscriptionsVersion { get; set; }

src/Abc.Zebus/SubscriptionRequestBatch.cs

@@ -2,14 +2,13 @@
 using System.Collections.Generic;
 using System.Linq;
 using System.Threading.Tasks;
-using JetBrains.Annotations;
 
 namespace Abc.Zebus
 {
     public class SubscriptionRequestBatch
     {
-        private readonly TaskCompletionSource<object> _submitCompletionSource = new TaskCompletionSource<object>();
-        private readonly TaskCompletionSource<object> _registerCompletionSource = new TaskCompletionSource<object>();
+        private readonly TaskCompletionSource<object?> _submitCompletionSource = new TaskCompletionSource<object?>();
+        private readonly TaskCompletionSource<object?> _registerCompletionSource = new TaskCompletionSource<object?>();
         private readonly List<SubscriptionRequest> _requests = new List<SubscriptionRequest>();
         private bool _isConsumed;
 
@@ -41,7 +40,7 @@ namespace Abc.Zebus
         internal Task WhenSubmittedAsync()
             => _submitCompletionSource.Task;
 
-        internal void NotifyRegistrationCompleted(Exception exception)
+        internal void NotifyRegistrationCompleted(Exception? exception)
         {
             if (exception != null)
                 _registerCompletionSource.SetException(exception);
@@ -60,8 +59,7 @@ namespace Abc.Zebus
             await _registerCompletionSource.Task.ConfigureAwait(false);
         }
 
-        [CanBeNull]
-        internal IEnumerable<Subscription> TryConsumeBatchSubscriptions()
+        internal IEnumerable<Subscription>? TryConsumeBatchSubscriptions()
         {
             lock (_requests)
             {

src/Abc.Zebus/Transport/OriginatorInfo.cs

@@ -14,12 +14,12 @@ namespace Abc.Zebus.Transport
         public readonly string SenderEndPoint;
 
         [ProtoMember(3, IsRequired = true)]
-        internal readonly string SenderMachineName;
+        internal readonly string? SenderMachineName;
 
         [ProtoMember(5, IsRequired = true)]
-        public string InitiatorUserName;
+        public string? InitiatorUserName;
 
-        public OriginatorInfo(PeerId senderId, string senderEndPoint, string senderMachineName, string initiatorUserName)
+        public OriginatorInfo(PeerId senderId, string senderEndPoint, string? senderMachineName, string? initiatorUserName)
         {
             SenderId = senderId;
             SenderEndPoint = senderEndPoint;
@@ -30,6 +30,7 @@ namespace Abc.Zebus.Transport
         [UsedImplicitly]
         internal OriginatorInfo()
         {
+            SenderEndPoint = default!;
         }
 
         public string GetSenderMachineNameFromEndPoint()

src/Abc.Zebus/Transport/SendContext.cs

@@ -5,7 +5,7 @@ namespace Abc.Zebus.Transport
     public class SendContext
     {
         public List<PeerId> PersistentPeerIds { get; } = new List<PeerId>();
-        public Peer PersistencePeer { get; set; }
+        public Peer? PersistencePeer { get; set; }
 
         public bool WasPersisted(PeerId peerId)
         {

src/Abc.Zebus/Transport/TransportMessage.cs

@@ -24,19 +24,19 @@ namespace Abc.Zebus.Transport
         }
 
         [ProtoIgnore, JsonIgnore]
-        public Stream Content { get; set; }
+        public Stream? Content { get; set; }
 
         [ProtoMember(4, IsRequired = true)]
         public OriginatorInfo Originator { get; set; }
 
         [ProtoMember(5, IsRequired = false)]
-        public string Environment { get; set; }
+        public string? Environment { get; set; }
 
         [ProtoMember(6, IsRequired = false)]
         public bool? WasPersisted { get; set; }
 
         [ProtoMember(7, IsRequired = false)]
-        public List<PeerId> PersistentPeerIds { get; set; }
+        public List<PeerId>? PersistentPeerIds { get; set; }
 
         [JsonIgnore]
         public bool IsPersistTransportMessage => PersistentPeerIds != null && PersistentPeerIds.Count != 0;
@@ -65,6 +65,7 @@ namespace Abc.Zebus.Transport
         [UsedImplicitly]
         internal TransportMessage()
         {
+            Originator = default!;
         }
 
         private static OriginatorInfo CreateOriginator(PeerId peerId, string peerEndPoint)
@@ -89,7 +90,7 @@ namespace Abc.Zebus.Transport
         internal TransportMessage ToPersistTransportMessage(List<PeerId> peerIds) => CloneWithPeerIds(peerIds);
         internal TransportMessage UnpackPersistTransportMessage() => CloneWithPeerIds(null);
 
-        private TransportMessage CloneWithPeerIds(List<PeerId> peerIds) => new TransportMessage
+        private TransportMessage CloneWithPeerIds(List<PeerId>? peerIds) => new TransportMessage
         {
             Id = Id,
             MessageTypeId = MessageTypeId,

src/Abc.Zebus/Transport/TransportMessageReader.cs

@@ -52,8 +52,8 @@ namespace Abc.Zebus.Transport
             var endPosition = input.Position + length;
 
             var senderId = new PeerId();
-            string senderEndPoint = null;
-            string initiatorUserName = null;
+            string? senderEndPoint = null;
+            string? initiatorUserName = null;
 
             while (input.Position < endPosition && input.TryReadTag(out var number, out var wireType))
             {
@@ -74,7 +74,7 @@ namespace Abc.Zebus.Transport
                 }
             }
 
-            return new OriginatorInfo(senderId, senderEndPoint, null, initiatorUserName);
+            return new OriginatorInfo(senderId, senderEndPoint!, null, initiatorUserName);
         }
 
         private static PeerId ReadPeerId(CodedInputStream input)
@@ -106,7 +106,7 @@ namespace Abc.Zebus.Transport
             var length = input.ReadLength();
             var endPosition = input.Position + length;
 
-            var value = default(T);
+            var value = default(T)!;
 
             while (input.Position < endPosition && input.TryReadTag(out var number, out var wireType))
             {
@@ -124,7 +124,7 @@ namespace Abc.Zebus.Transport
             return value;
         }
 
-        private static List<PeerId> ReadPeerIds(CodedInputStream input, List<PeerId> peerIds)
+        private static List<PeerId> ReadPeerIds(CodedInputStream input, List<PeerId>? peerIds)
         {
             if (peerIds == null)
                 peerIds = new List<PeerId>();

src/Abc.Zebus/Transport/TransportMessageWriter.cs

@@ -1,5 +1,4 @@
-using System;
-using System.Collections.Generic;
+using System.Collections.Generic;
 using System.IO;
 using Abc.Zebus.Serialization.Protobuf;
 
@@ -9,7 +8,7 @@ namespace Abc.Zebus.Transport
     {
         private static readonly MemoryStream _emptyStream = new MemoryStream(new byte[0]);
 
-        internal static void WriteTransportMessage(this CodedOutputStream output, TransportMessage transportMessage, string environmentOverride = null)
+        internal static void WriteTransportMessage(this CodedOutputStream output, TransportMessage transportMessage, string? environmentOverride = null)
         {
             output.WriteRawTag(1 << 3 | 2);
             Write(output, transportMessage.Id);
@@ -45,7 +44,7 @@ namespace Abc.Zebus.Transport
             WriteWasPersisted(output, wasPersisted);
         }
 
-        internal static void WritePersistentPeerIds(this CodedOutputStream output, TransportMessage transportMessage, List<PeerId> persistentPeerIdOverride)
+        internal static void WritePersistentPeerIds(this CodedOutputStream output, TransportMessage transportMessage, List<PeerId>? persistentPeerIdOverride)
         {
             var peerIds = persistentPeerIdOverride ?? transportMessage.PersistentPeerIds;
             if (peerIds == null)

src/Abc.Zebus/Transport/Zmq/ZmqUtil.cs

@@ -1,4 +1,5 @@
 using System;
+using System.Diagnostics.CodeAnalysis;
 using System.IO;
 using System.Runtime.CompilerServices;
 using System.Runtime.InteropServices;
@@ -8,7 +9,7 @@ namespace Abc.Zebus.Transport.Zmq
 {
     internal static unsafe class ZmqUtil
     {
-        [ContractAnnotation("=> halt")]
+        [ContractAnnotation("=> halt"), DoesNotReturn]
         [MethodImpl(MethodImplOptions.NoInlining)]
         public static Exception ThrowLastError(string message)
             => throw new IOException($"{message}: {GetLastErrorMessage()})");
@@ -31,6 +32,7 @@ namespace Abc.Zebus.Transport.Zmq
         public static bool WasInterrupted()
             => ZmqNative.errno() == ZmqErrorCode.EINTR;
 
+        [ContractAnnotation("=> halt"), DoesNotReturn]
         [MethodImpl(MethodImplOptions.NoInlining)]
         public static void ThrowArgOutOfRange()
             => throw new ArgumentOutOfRangeException();

src/Abc.Zebus/Transport/ZmqInboundSocket.cs

@@ -16,8 +16,8 @@ namespace Abc.Zebus.Transport
         private readonly ZmqSocketOptions _options;
         private readonly string _environment;
         private byte[] _readBuffer = new byte[0];
-        private ZmqSocket _socket;
-        private ZmqEndPoint _socketEndPoint;
+        private ZmqSocket? _socket;
+        private ZmqEndPoint? _socketEndPoint;
         private TimeSpan _lastReceiveTimeout;
 
         public ZmqInboundSocket(ZmqContext context, PeerId peerId, ZmqEndPoint originalEndpoint, ZmqSocketOptions options, string environment)
@@ -47,19 +47,19 @@ namespace Abc.Zebus.Transport
 
         public void Dispose()
         {
-            _socket.Dispose();
+            _socket?.Dispose();
         }
 
-        public CodedInputStream Receive(TimeSpan? timeout = null)
+        public CodedInputStream? Receive(TimeSpan? timeout = null)
         {
             var receiveTimeout = timeout ?? _options.ReceiveTimeout;
             if (receiveTimeout != _lastReceiveTimeout)
             {
-                _socket.SetOption(ZmqSocketOption.RCVTIMEO, (int)receiveTimeout.TotalMilliseconds);
+                _socket!.SetOption(ZmqSocketOption.RCVTIMEO, (int)receiveTimeout.TotalMilliseconds);
                 _lastReceiveTimeout = receiveTimeout;
             }
 
-            if (_socket.TryReadMessage(ref _readBuffer, out var messageLength, out var error))
+            if (_socket!.TryReadMessage(ref _readBuffer, out var messageLength, out var error))
                 return new CodedInputStream(_readBuffer, 0, messageLength);
 
             // EAGAIN: Non-blocking mode was requested and no messages are available at the moment.
@@ -83,12 +83,12 @@ namespace Abc.Zebus.Transport
 
         public void Disconnect()
         {
-            var endpoint = _socket.GetOptionString(ZmqSocketOption.LAST_ENDPOINT);
+            var endpoint = _socket?.GetOptionString(ZmqSocketOption.LAST_ENDPOINT);
             if (endpoint == null)
                 return;
 
             _logger.InfoFormat("Unbinding socket, Inbound Endpoint: {0}", endpoint);
-            if (!_socket.TryUnbind(endpoint))
+            if (!_socket!.TryUnbind(endpoint))
                 _logger.WarnFormat("Socket error, Inbound Endpoint: {0}, Error: {1}", endpoint, ZmqUtil.GetLastErrorMessage());
         }
     }

src/Abc.Zebus/Transport/ZmqOutboundSocket.cs

@@ -14,7 +14,7 @@ namespace Abc.Zebus.Transport
         private readonly ZmqContext _context;
         private readonly ZmqSocketOptions _options;
         private readonly IZmqOutboundSocketErrorHandler _errorHandler;
-        private ZmqSocket _socket;
+        private ZmqSocket? _socket;
         private int _failedSendCount;
         private bool _isInClosedState;
         private TimeSpan _closedStateDuration;
@@ -48,7 +48,7 @@ namespace Abc.Zebus.Transport
             }
             catch (Exception ex)
             {
-                _socket.Dispose();
+                _socket?.Dispose();
                 _socket = null;
                 IsConnected = false;
 
@@ -96,8 +96,8 @@ namespace Abc.Zebus.Transport
 
             try
             {
-                _socket.SetOption(ZmqSocketOption.LINGER, 0);
-                _socket.Dispose();
+                _socket!.SetOption(ZmqSocketOption.LINGER, 0);
+                _socket!.Dispose();
 
                 _logger.InfoFormat("Socket disconnected, Peer: {0}", PeerId);
             }
@@ -115,7 +115,7 @@ namespace Abc.Zebus.Transport
             if (!CanSendOrConnect(message))
                 return;
 
-            if (_socket.TrySend(buffer, 0, length, out var error))
+            if (_socket!.TrySend(buffer, 0, length, out var error))
             {
                 _failedSendCount = 0;
                 return;

src/Abc.Zebus/Transport/ZmqTransport.cs

@@ -18,17 +18,17 @@ namespace Abc.Zebus.Transport
         private readonly IZmqOutboundSocketErrorHandler _errorHandler;
         private readonly ZmqEndPoint _configuredInboundEndPoint;
         private ILog _logger = LogManager.GetLogger(typeof(ZmqTransport));
-        private ConcurrentDictionary<PeerId, ZmqOutboundSocket> _outboundSockets;
-        private BlockingCollection<OutboundSocketAction> _outboundSocketActions;
-        private BlockingCollection<PendingDisconnect> _pendingDisconnects;
-        private ZmqContext _context;
-        private Thread _inboundThread;
-        private Thread _outboundThread;
-        private Thread _disconnectThread;
+        private ConcurrentDictionary<PeerId, ZmqOutboundSocket> _outboundSockets = new ConcurrentDictionary<PeerId, ZmqOutboundSocket>();
+        private BlockingCollection<OutboundSocketAction>? _outboundSocketActions;
+        private BlockingCollection<PendingDisconnect> _pendingDisconnects = new BlockingCollection<PendingDisconnect>();
+        private ZmqContext? _context;
+        private Thread? _inboundThread;
+        private Thread? _outboundThread;
+        private Thread? _disconnectThread;
         private volatile bool _isListening;
-        private ZmqEndPoint _realInboundEndPoint;
-        private string _environment;
-        private CountdownEvent _outboundSocketsToStop;
+        private ZmqEndPoint? _realInboundEndPoint;
+        private string _environment = string.Empty;
+        private CountdownEvent? _outboundSocketsToStop;
         private bool _isRunning;
 
         public ZmqTransport(IZmqTransportConfiguration configuration, ZmqSocketOptions socketOptions, IZmqOutboundSocketErrorHandler errorHandler)
@@ -39,7 +39,7 @@ namespace Abc.Zebus.Transport
             SocketOptions = socketOptions;
         }
 
-        public event Action<TransportMessage> MessageReceived;
+        public event Action<TransportMessage>? MessageReceived;
 
         public bool IsListening => _isListening;
 
@@ -121,25 +121,25 @@ namespace Abc.Zebus.Transport
             if (discardPendingMessages)
                 DiscardItems(_pendingDisconnects);
 
-            if (!_disconnectThread.Join(30.Seconds()))
+            if (!_disconnectThread!.Join(30.Seconds()))
                 _logger.Error("Unable to terminate disconnect thread");
 
-            _outboundSocketActions.CompleteAdding();
+            _outboundSocketActions!.CompleteAdding();
 
             if (discardPendingMessages)
-                DiscardItems(_outboundSocketActions);
+                DiscardItems(_outboundSocketActions!);
 
-            if (!_outboundThread.Join(30.Seconds()))
+            if (!_outboundThread!.Join(30.Seconds()))
                 _logger.Error("Unable to terminate outbound thread");
 
             _isListening = false;
-            if (!_inboundThread.Join(30.Seconds()))
+            if (!_inboundThread!.Join(30.Seconds()))
                 _logger.Error("Unable to terminate inbound thread");
 
-            _outboundSocketActions.Dispose();
+            _outboundSocketActions!.Dispose();
             _outboundSocketActions = null;
 
-            _context.Dispose();
+            _context!.Dispose();
             _logger.InfoFormat("{0} Stopped", PeerId);
         }
 
@@ -157,7 +157,7 @@ namespace Abc.Zebus.Transport
 
         public void Send(TransportMessage message, IEnumerable<Peer> peers, SendContext context)
         {
-            _outboundSocketActions.Add(OutboundSocketAction.Send(message, peers, context));
+            _outboundSocketActions!.Add(OutboundSocketAction.Send(message, peers, context));
         }
 
         private void Disconnect(PeerId peerId, bool delayed = true)
@@ -171,7 +171,7 @@ namespace Abc.Zebus.Transport
             }
             else
             {
-                SafeAdd(_outboundSocketActions, OutboundSocketAction.Disconnect(peerId));
+                SafeAdd(_outboundSocketActions!, OutboundSocketAction.Disconnect(peerId));
             }
         }
 
@@ -205,12 +205,12 @@ namespace Abc.Zebus.Transport
             _logger.InfoFormat("InboundProc terminated");
         }
 
-        private ZmqInboundSocket CreateInboundSocket(InboundProcStartSequenceState state)
+        private ZmqInboundSocket? CreateInboundSocket(InboundProcStartSequenceState state)
         {
-            ZmqInboundSocket inboundSocket = null;
+            ZmqInboundSocket? inboundSocket = null;
             try
             {
-                inboundSocket = new ZmqInboundSocket(_context, PeerId, _configuredInboundEndPoint, SocketOptions, _environment);
+                inboundSocket = new ZmqInboundSocket(_context!, PeerId, _configuredInboundEndPoint, SocketOptions, _environment);
                 _realInboundEndPoint = inboundSocket.Bind();
                 return inboundSocket;
             }
@@ -231,7 +231,7 @@ namespace Abc.Zebus.Transport
         {
             inboundSocket.Disconnect();
 
-            CodedInputStream inputStream;
+            CodedInputStream? inputStream;
             while ((inputStream = inboundSocket.Receive(100.Milliseconds())) != null)
                 DeserializeAndForwardTransportMessage(inputStream);
         }
@@ -279,7 +279,7 @@ namespace Abc.Zebus.Transport
 
             _logger.InfoFormat("Received EndOfStreamAck for {0}, {1}", senderId, senderEndPoint);
 
-            _outboundSocketsToStop.Signal();
+            _outboundSocketsToStop!.Signal();
         }
 
         private void SendEndOfStreamAck(TransportMessage transportMessage)
@@ -289,8 +289,8 @@ namespace Abc.Zebus.Transport
             var endOfStreamAck = new TransportMessage(MessageTypeId.EndOfStreamAck, new MemoryStream(), PeerId, InboundEndPoint);
             var closingPeer = new Peer(transportMessage.Originator.SenderId, transportMessage.Originator.SenderEndPoint);
 
-            SafeAdd(_outboundSocketActions, OutboundSocketAction.Send(endOfStreamAck, new[] { closingPeer }, new SendContext()));
-            SafeAdd(_pendingDisconnects, new PendingDisconnect(closingPeer.Id, SystemDateTime.UtcNow.Add(_configuration.WaitForEndOfStreamAckTimeout)));
+            SafeAdd(_outboundSocketActions!, OutboundSocketAction.Send(endOfStreamAck, new[] { closingPeer }, new SendContext()));
+            SafeAdd(_pendingDisconnects!, new PendingDisconnect(closingPeer.Id, SystemDateTime.UtcNow.Add(_configuration.WaitForEndOfStreamAckTimeout)));
         }
 
         private bool IsFromCurrentEnvironment(TransportMessage transportMessage)
@@ -315,7 +315,7 @@ namespace Abc.Zebus.Transport
 
             var outputStream = new CodedOutputStream();
 
-            foreach (var socketAction in _outboundSocketActions.GetConsumingEnumerable())
+            foreach (var socketAction in _outboundSocketActions!.GetConsumingEnumerable())
             {
                 if (socketAction.IsDisconnect)
                 {
@@ -389,7 +389,7 @@ namespace Abc.Zebus.Transport
         {
             if (!_outboundSockets.TryGetValue(peer.Id, out var outboundSocket))
             {
-                outboundSocket = new ZmqOutboundSocket(_context, peer.Id, peer.EndPoint, SocketOptions, _errorHandler);
+                outboundSocket = new ZmqOutboundSocket(_context!, peer.Id, peer.EndPoint, SocketOptions, _errorHandler);
                 outboundSocket.ConnectFor(transportMessage);
 
                 _outboundSockets.TryAdd(peer.Id, outboundSocket);
@@ -444,7 +444,7 @@ namespace Abc.Zebus.Transport
                     Thread.Sleep(500);
                 }
 
-                SafeAdd(_outboundSocketActions, OutboundSocketAction.Disconnect(pendingDisconnect.PeerId));
+                SafeAdd(_outboundSocketActions!, OutboundSocketAction.Disconnect(pendingDisconnect.PeerId));
             }
         }
 
@@ -462,7 +462,7 @@ namespace Abc.Zebus.Transport
 
         private readonly struct OutboundSocketAction
         {
-            private static readonly TransportMessage _disconnectMessage = new TransportMessage(default, null, new PeerId(), null);
+            private static readonly TransportMessage _disconnectMessage = new TransportMessage(default, null!, new PeerId(), null!);
 
             private OutboundSocketAction(TransportMessage message, IEnumerable<Peer> targets, SendContext context)
             {
@@ -480,7 +480,7 @@ namespace Abc.Zebus.Transport
                 => new OutboundSocketAction(message, peers, context);
 
             public static OutboundSocketAction Disconnect(PeerId peerId)
-                => new OutboundSocketAction(_disconnectMessage, new List<Peer> { new Peer(peerId, null) }, null);
+                => new OutboundSocketAction(_disconnectMessage, new List<Peer> { new Peer(peerId, null!) }, null!);
         }
 
         private class PendingDisconnect
@@ -497,7 +497,7 @@ namespace Abc.Zebus.Transport
 
         private class InboundProcStartSequenceState
         {
-            private Exception _inboundProcStartException;
+            private Exception? _inboundProcStartException;
             private readonly ManualResetEvent _inboundProcStartedSignal = new ManualResetEvent(false);
 
             public void Wait()

src/Abc.Zebus/Util/BackgroundThread.cs

@@ -8,7 +8,7 @@ namespace Abc.Zebus.Util
     {
         private static readonly ILog _logger = LogManager.GetLogger(typeof(BackgroundThread));
 
-        public static Thread Start(ThreadStart startAction, Action abortAction = null)
+        public static Thread Start(ThreadStart startAction, Action? abortAction = null)
         {
             var thread = new Thread(Wrapper(startAction, abortAction))
             {
@@ -19,7 +19,7 @@ namespace Abc.Zebus.Util
             return thread;
         }
 
-        public static Thread Start<T>(Action<T> startAction, T state, Action abortAction = null)
+        public static Thread Start<T>(Action<T> startAction, T state, Action? abortAction = null)
         {
             var thread = new Thread(Wrapper(startAction, abortAction))
             {
@@ -42,7 +42,7 @@ namespace Abc.Zebus.Util
             }
         }
 
-        private static ParameterizedThreadStart Wrapper<T>(Action<T> action, Action abortAction)
+        private static ParameterizedThreadStart Wrapper<T>(Action<T> action, Action? abortAction)
         {
             return s =>
             {
@@ -64,7 +64,7 @@ namespace Abc.Zebus.Util
             };
         }
 
-        private static ThreadStart Wrapper(ThreadStart action, Action abortAction)
+        private static ThreadStart Wrapper(ThreadStart action, Action? abortAction)
         {
             return () =>
             {

src/Abc.Zebus/Util/Collections/ConcurrentSet.cs

@@ -5,12 +5,12 @@ using System.Collections.Generic;
 namespace Abc.Zebus.Util.Collections
 {
     internal class ConcurrentSet<T> : ICollection<T>
+        where T : notnull
     {
-        private readonly ConcurrentDictionary<T, object> _items;
+        private readonly ConcurrentDictionary<T, object?> _items = new ConcurrentDictionary<T, object?>();
 
         public ConcurrentSet()
         {
-            _items = new ConcurrentDictionary<T, object>();
         }
 
         public ConcurrentSet(IEnumerable<T> items)
@@ -61,8 +61,7 @@ namespace Abc.Zebus.Util.Collections
 
         public bool Remove(T item)
         {
-            object value;
-            return _items.TryRemove(item, out value);
+            return _items.TryRemove(item, out var value);
         }
     }
 }

src/Abc.Zebus/Util/Collections/FlushableBlockingCollection.cs

@@ -10,7 +10,7 @@ namespace Abc.Zebus.Util.Collections
         private readonly ManualResetEventSlim _addSignal = new ManualResetEventSlim();
         private volatile ConcurrentQueue<T> _queue = new ConcurrentQueue<T>();
         private volatile bool _isAddingCompleted;
-        private ManualResetEventSlim _isEmptySignal;
+        private ManualResetEventSlim? _isEmptySignal;
         private int _hasChangedSinceLastWaitForEmpty;
 
         public int Count => _queue.Count;

src/Abc.Zebus/Util/DisposableAction.cs

@@ -1,4 +1,4 @@
-// Copyright (c) Lokad 2009 
+// Copyright (c) Lokad 2009
 // https://github.com/Lokad/lokad-shared-libraries
 // This code is released under the terms of the new BSD licence
 
@@ -13,7 +13,7 @@ namespace Abc.Zebus.Util
     [Serializable]
     internal sealed class DisposableAction : IDisposable
     {
-        private Action _action;
+        private Action? _action;
 
         /// <summary>
         /// Initializes a new instance of the <see cref="DisposableAction"/> class.

src/Abc.Zebus/Util/Extensions/ExtendDictionary.cs

@@ -1,6 +1,7 @@
 using System;
 using System.Collections.Concurrent;
 using System.Collections.Generic;
+using System.Diagnostics.CodeAnalysis;
 using JetBrains.Annotations;
 
 namespace Abc.Zebus.Util.Extensions
@@ -8,11 +9,13 @@ namespace Abc.Zebus.Util.Extensions
     internal static class ExtendDictionary
     {
         public static TValue GetValueOrAdd<TKey, TValue>(this IDictionary<TKey, TValue> dictionary, TKey key, Func<TValue> valueBuilder)
+            where TKey : notnull
         {
             return dictionary.GetValueOrAdd(key, _ => valueBuilder());
         }
 
         public static TValue GetValueOrAdd<TKey, TValue>(this IDictionary<TKey, TValue> dictionary, TKey key, Func<TKey, TValue> valueBuilder)
+            where TKey : notnull
         {
             if (!dictionary.TryGetValue(key, out var value))
             {
@@ -25,39 +28,49 @@ namespace Abc.Zebus.Util.Extensions
 
         [Pure]
         public static TValue? GetValueOrNull<TKey, TValue>(this IDictionary<TKey, TValue> dictionary, TKey key)
+            where TKey : notnull
             where TValue : struct
         {
             return dictionary.TryGetValue(key, out var value) ? value : (TValue?)null;
         }
 
         [Pure]
+        [return: MaybeNull]
         public static TValue GetValueOrDefault<TKey, TValue>(this IDictionary<TKey, TValue> dictionary, TKey key)
+            where TKey : notnull
         {
-            return GetValueOrDefault(dictionary, key, default(TValue));
+            return GetValueOrDefault(dictionary, key, default(TValue)!);
         }
 
         [Pure]
+        [return: MaybeNull]
+        [return: NotNullIfNotNull("defaultValue")]
         public static TValue GetValueOrDefault<TKey, TValue>(this IDictionary<TKey, TValue> dictionary, TKey key, TValue defaultValue)
+            where TKey : notnull
         {
             return dictionary.TryGetValue(key, out var value) ? value : defaultValue;
         }
 
         public static TValue GetValueOrDefault<TKey, TValue>(this IDictionary<TKey, TValue> dictionary, TKey key, [InstantHandle] Func<TKey, TValue> defaultValueBuilder)
+            where TKey : notnull
         {
             return dictionary.TryGetValue(key, out var value) ? value : defaultValueBuilder(key);
         }
 
         public static bool Remove<TKey, TValue>(this ConcurrentDictionary<TKey, TValue> dictionary, TKey key)
+            where TKey : notnull
         {
             return dictionary.TryRemove(key, out _);
         }
 
         public static bool TryRemove<TKey, TValue>(this ConcurrentDictionary<TKey, TValue> dictionary, TKey key, TValue comparisonValue)
+            where TKey : notnull
         {
             return ((ICollection<KeyValuePair<TKey, TValue>>)dictionary).Remove(new KeyValuePair<TKey, TValue>(key, comparisonValue));
         }
 
         public static void RemoveRange<TKey, TValue>(this IDictionary<TKey, TValue> dictionary, IEnumerable<TKey> keys)
+            where TKey : notnull
         {
             foreach (var key in keys)
             {
@@ -66,6 +79,7 @@ namespace Abc.Zebus.Util.Extensions
         }
 
         public static void Deconstruct<TKey, TValue>(this KeyValuePair<TKey, TValue> pair, out TKey key, out TValue value)
+            where TKey : notnull
         {
             key = pair.Key;
             value = pair.Value;

src/Abc.Zebus/Util/Extensions/ExtendEnum.cs

@@ -8,16 +8,16 @@ namespace Abc.Zebus.Util.Extensions
         public static string GetAttributeDescription(this Enum enumValue)
         {
             var attribute = enumValue.GetAttributeOfType<DescriptionAttribute>();
-
-            return attribute == null ? String.Empty : attribute.Description;
+            return attribute == null ? string.Empty : attribute.Description;
         }
 
-        private static T GetAttributeOfType<T>(this Enum enumVal) where T : Attribute
+        private static T? GetAttributeOfType<T>(this Enum enumVal)
+            where T : Attribute
         {
             var enumType = enumVal.GetType();
             var memberInfo = enumType.GetMember(enumVal.ToString());
 
-            return memberInfo[0].GetAttribute<T>(false);
+            return memberInfo?[0].GetAttribute<T>(false);
         }
     }
 }

src/Abc.Zebus/Util/Extensions/ExtendIEnumerable.cs

@@ -7,7 +7,6 @@ namespace Abc.Zebus.Util.Extensions
 {
     internal static class ExtendIEnumerable
     {
-#if !NETFWK
         [Pure]
         public static HashSet<T> ToHashSet<T>([InstantHandle] this IEnumerable<T> collection)
         {
@@ -19,7 +18,6 @@ namespace Abc.Zebus.Util.Extensions
         {
             return new HashSet<T>(collection, comparer);
         }
-#endif
 
         [Pure]
         public static IEnumerable<T> Shuffle<T>(this IEnumerable<T> source)

src/Abc.Zebus/Util/Extensions/ExtendString.cs

@@ -1,17 +1,20 @@
+using System.Diagnostics.CodeAnalysis;
+
 namespace Abc.Zebus.Util.Extensions
 {
     internal static class ExtendString
     {
-        public static string Qualifier(this string input)
+        [return: NotNullIfNotNull("input")]
+        public static string? Qualifier(this string? input)
         {
             if (input == null)
                 return null;
-            
+
             var lastDotIndex = input.LastIndexOf('.');
             if (lastDotIndex == -1)
                 return input;
-            
+
             return input.Substring(0, lastDotIndex);
         }
     }
-}
+}

src/Abc.Zebus/Util/Extensions/ExtendTask.cs

@@ -7,7 +7,7 @@ namespace Abc.Zebus.Util.Extensions
 {
     internal static class ExtendTask
     {
-        public static async Task WithTimeoutAsync(this Task task, TimeSpan timeout, CancellationToken cancellationToken = default(CancellationToken))
+        public static async Task WithTimeoutAsync(this Task task, TimeSpan timeout, CancellationToken cancellationToken = default)
         {
             if (task.Status == TaskStatus.RanToCompletion)
                 return;
@@ -22,7 +22,7 @@ namespace Abc.Zebus.Util.Extensions
             await task.ConfigureAwait(false);
         }
 
-        public static async Task<T> WithTimeoutAsync<T>(this Task<T> task, TimeSpan timeout, CancellationToken cancellationToken = default(CancellationToken))
+        public static async Task<T> WithTimeoutAsync<T>(this Task<T> task, TimeSpan timeout, CancellationToken cancellationToken = default)
         {
             if (task.Status == TaskStatus.RanToCompletion)
                 return task.Result;
@@ -45,7 +45,9 @@ namespace Abc.Zebus.Util.Extensions
             }
             catch (AggregateException ex)
             {
-                ExceptionDispatchInfo.Capture(ex.InnerException).Throw();
+                if (ex.InnerException != null)
+                    ExceptionDispatchInfo.Capture(ex.InnerException).Throw();
+
                 throw;
             }
         }

src/Abc.Zebus/Util/Extensions/ExtendType.cs

@@ -1,6 +1,6 @@
 #region (c)2009 Lokad - New BSD license
 
-// Copyright (c) Lokad 2009 
+// Copyright (c) Lokad 2009
 // Company: http://www.lokad.com
 // This code is released under the terms of the new BSD licence
 
@@ -25,7 +25,8 @@ namespace Abc.Zebus.Util.Extensions
         ///<param name="inherit"><see cref="MemberInfo.GetCustomAttributes(Type,bool)"/></param>
         /// <returns><em>Null</em> if the attribute is not found</returns>
         /// <exception cref="InvalidOperationException">If there are 2 or more attributes</exception>
-        public static T GetAttribute<T>(this ICustomAttributeProvider target, bool inherit) where T : Attribute
+        public static T? GetAttribute<T>(this ICustomAttributeProvider target, bool inherit)
+            where T : Attribute
         {
             if (target.IsDefined(typeof (T), inherit))
             {
@@ -36,6 +37,7 @@ namespace Abc.Zebus.Util.Extensions
                 }
                 return (T) attributes[0];
             }
+
             return null;
         }
 

src/Abc.Zebus/Util/NullableAnnotations.cs

@@ -0,0 +1,72 @@
+// ReSharper disable CheckNamespace
+namespace System.Diagnostics.CodeAnalysis
+{
+    [AttributeUsageAttribute(AttributeTargets.Field | AttributeTargets.Parameter | AttributeTargets.Property)]
+    internal sealed class AllowNullAttribute : Attribute
+    {
+    }
+
+    [AttributeUsageAttribute(AttributeTargets.Field | AttributeTargets.Parameter | AttributeTargets.Property)]
+    internal sealed class DisallowNullAttribute : Attribute
+    {
+    }
+
+    [AttributeUsageAttribute(AttributeTargets.Method, Inherited = false)]
+    internal sealed class DoesNotReturnAttribute : Attribute
+    {
+    }
+
+    [AttributeUsageAttribute(AttributeTargets.Parameter)]
+    internal sealed class DoesNotReturnIfAttribute : Attribute
+    {
+        public DoesNotReturnIfAttribute(bool parameterValue)
+        {
+            ParameterValue = parameterValue;
+        }
+
+        public bool ParameterValue { get; }
+    }
+
+    [AttributeUsageAttribute(AttributeTargets.Field | AttributeTargets.Parameter | AttributeTargets.Property | AttributeTargets.ReturnValue)]
+    internal sealed class MaybeNullAttribute : Attribute
+    {
+    }
+
+    [AttributeUsageAttribute(AttributeTargets.Parameter)]
+    internal sealed class MaybeNullWhenAttribute : Attribute
+    {
+        public MaybeNullWhenAttribute(bool returnValue)
+        {
+            ReturnValue = returnValue;
+        }
+
+        public bool ReturnValue { get; }
+    }
+
+    [AttributeUsageAttribute(AttributeTargets.Field | AttributeTargets.Parameter | AttributeTargets.Property | AttributeTargets.ReturnValue)]
+    internal sealed class NotNullAttribute : Attribute
+    {
+    }
+
+    [AttributeUsageAttribute(AttributeTargets.Parameter | AttributeTargets.Property | AttributeTargets.ReturnValue, AllowMultiple = true)]
+    internal sealed class NotNullIfNotNullAttribute : Attribute
+    {
+        public NotNullIfNotNullAttribute(string parameterName)
+        {
+            ParameterName = parameterName;
+        }
+
+        public string ParameterName { get; }
+    }
+
+    [AttributeUsageAttribute(AttributeTargets.Parameter)]
+    internal sealed class NotNullWhenAttribute : Attribute
+    {
+        public NotNullWhenAttribute(bool returnValue)
+        {
+            ReturnValue = returnValue;
+        }
+
+        public bool ReturnValue { get; }
+    }
+}

src/Abc.Zebus/Util/SystemDateTime.cs

@@ -24,8 +24,8 @@ namespace Abc.Zebus.Util
             if (now == null && utcNow == null)
                 throw new ArgumentNullException();
 
-            _nowFunc = () => now ?? utcNow.Value.ToLocalTime();
-            _utcNowFunc = () => utcNow ?? now.Value.ToUniversalTime();
+            _nowFunc = () => now ?? utcNow.GetValueOrDefault().ToLocalTime();
+            _utcNowFunc = () => utcNow ?? now.GetValueOrDefault().ToUniversalTime();
             return new Scope();
         }
 
@@ -43,4 +43,4 @@ namespace Abc.Zebus.Util
             }
         }
     }
-}
+}

src/Abc.Zebus/Util/TypeUtil.cs

@@ -9,14 +9,14 @@ namespace Abc.Zebus.Util
 {
     internal static class TypeUtil
     {
-        private static readonly ConcurrentDictionary<string, Type> _typesByNames = new ConcurrentDictionary<string, Type>();
+        private static readonly ConcurrentDictionary<string, Type?> _typesByNames = new ConcurrentDictionary<string, Type?>();
 
-        public static Type Resolve(string typeName)
+        public static Type? Resolve(string typeName)
         {
             return _typesByNames.GetOrAdd(typeName, FindTypeByName);
         }
-        
-        private static Type FindTypeByName(string typeName)
+
+        private static Type? FindTypeByName(string typeName)
         {
             try
             {
@@ -30,7 +30,7 @@ namespace Abc.Zebus.Util
 
             if (typeName.Contains("<"))
                 return FindGenericTypeByName(typeName);
-            
+
             foreach (var assembly in AppDomain.CurrentDomain.GetAssemblies().OrderBy(a => a.FullName))
             {
                 var type = assembly.GetType(typeName);
@@ -41,7 +41,7 @@ namespace Abc.Zebus.Util
             return null;
         }
 
-        private static Type FindGenericTypeByName(string typeName)
+        private static Type? FindGenericTypeByName(string typeName)
         {
             var genericArguments = typeName.Substring(typeName.IndexOf("<", StringComparison.Ordinal)).Trim('<', '>').Split(new[] { ',' }, StringSplitOptions.RemoveEmptyEntries);
             var typeNameWithoutGenericArguments = typeName.Substring(0, typeName.IndexOf("<", StringComparison.Ordinal)) + '`' + genericArguments.Length;
@@ -49,7 +49,7 @@ namespace Abc.Zebus.Util
             var type = Resolve(typeNameWithoutGenericArguments);
             if (type == null)
                 return null;
-            
+
             var genericTypes = new List<Type>();
             foreach (var genericArgument in genericArguments)
             {
@@ -88,4 +88,4 @@ namespace Abc.Zebus.Util
             return builder.ToString();
         }
     }
-}
+}