Added identity observer

pawelpabich · pawelpabich · commit 755c1c27cad0 · 2025-03-10T13:53:31.000+10:00
diff --git a/source/Halibut.Tests/Transport/Protocol/ProtocolFixture.cs b/source/Halibut.Tests/Transport/Protocol/ProtocolFixture.cs
@@ -27,7 +27,7 @@ public void SetUp()
             stream.SetRemoteIdentity(new RemoteIdentity(RemoteIdentityType.Server));
             var limits = new HalibutTimeoutsAndLimitsForTestsBuilder().Build();
             var activeConnectionsLimiter = new ActiveTcpConnectionsLimiter(limits);
-            protocol = new MessageExchangeProtocol(stream, new HalibutTimeoutsAndLimitsForTestsBuilder().Build(), activeConnectionsLimiter, Substitute.For<ILog>());
+            protocol = new MessageExchangeProtocol(stream, new HalibutTimeoutsAndLimitsForTestsBuilder().Build(), activeConnectionsLimiter, NoIdentityObserver.Instance, Substitute.For<ILog>());
         }
 
         // TODO - ASYNC ME UP! ExchangeAsClientAsync cancellation
diff --git a/source/Halibut.Tests/Transport/SecureClientFixture.cs b/source/Halibut.Tests/Transport/SecureClientFixture.cs
@@ -67,7 +67,7 @@ public async Task SecureClientClearsPoolWhenAllConnectionsCorrupt()
                 var connection = Substitute.For<IConnection>();
                 var limits = new HalibutTimeoutsAndLimitsForTestsBuilder().Build();
                 var activeConnectionLimiter = new ActiveTcpConnectionsLimiter(limits);
-                connection.Protocol.Returns(new MessageExchangeProtocol(stream, limits, activeConnectionLimiter, log));
+                connection.Protocol.Returns(new MessageExchangeProtocol(stream, limits, activeConnectionLimiter, NoIdentityObserver.Instance, log));
 
                 await connectionManager.ReleaseConnectionAsync(endpoint, connection, CancellationToken.None);
             }
@@ -96,7 +96,7 @@ static MessageExchangeProtocol GetProtocol(Stream stream, ILog logger)
         {
             var limits = new HalibutTimeoutsAndLimitsForTestsBuilder().Build();
             var activeConnectionLimiter = new ActiveTcpConnectionsLimiter(limits);
-            return new MessageExchangeProtocol(new MessageExchangeStream(stream, new MessageSerializerBuilder(new LogFactory()).Build(), new NoOpControlMessageObserver(), limits, logger), limits, activeConnectionLimiter, logger);
+            return new MessageExchangeProtocol(new MessageExchangeStream(stream, new MessageSerializerBuilder(new LogFactory()).Build(), new NoOpControlMessageObserver(), limits, logger), limits, activeConnectionLimiter, NoIdentityObserver.Instance, logger);
         }
     }
 }
diff --git a/source/Halibut.Tests/Transport/SecureListenerFixture.cs b/source/Halibut.Tests/Transport/SecureListenerFixture.cs
@@ -73,7 +73,8 @@ public async Task SecureListenerDoesNotCreateHundredsOfIoEventsPerSecondOnWindow
                     (_, _) => UnauthorizedClientConnectResponse.BlockConnection,
                     timeoutsAndLimits,
                     new StreamFactory(),
-                    NoOpConnectionsObserver.Instance
+                    NoOpConnectionsObserver.Instance,
+                    NoIdentityObserver.Instance
                 );
 
                 var idleAverage = CollectCounterValues(opsPerSec)
diff --git a/source/Halibut/HalibutRuntime.cs b/source/Halibut/HalibutRuntime.cs
@@ -45,6 +45,7 @@ public class HalibutRuntime : IHalibutRuntime
         readonly IConnectionsObserver connectionsObserver;
         readonly IActiveTcpConnectionsLimiter activeTcpConnectionsLimiter;
         readonly IControlMessageObserver controlMessageObserver;
+        readonly IIdentityObserver identityObserver;
 
         internal HalibutRuntime(
             IServiceFactory serviceFactory,
@@ -59,7 +60,8 @@ internal HalibutRuntime(
             IStreamFactory streamFactory,
             IRpcObserver rpcObserver,
             IConnectionsObserver connectionsObserver, 
-            IControlMessageObserver controlMessageObserver)
+            IControlMessageObserver controlMessageObserver,
+            IIdentityObserver identityObserver)
         {
             this.serverCertificate = serverCertificate;
             this.trustProvider = trustProvider;
@@ -74,6 +76,7 @@ internal HalibutRuntime(
             TimeoutsAndLimits = halibutTimeoutsAndLimits;
             this.connectionsObserver = connectionsObserver;
             this.controlMessageObserver = controlMessageObserver;
+            this.identityObserver = identityObserver;
 
             connectionManager = new ConnectionManagerAsync();
             this.tcpConnectionFactory = new TcpConnectionFactory(serverCertificate, TimeoutsAndLimits, streamFactory);
@@ -106,7 +109,7 @@ public int Listen(int port)
 
         ExchangeProtocolBuilder ExchangeProtocolBuilder()
         {
-            return (stream, log) => new MessageExchangeProtocol(new MessageExchangeStream(stream, messageSerializer, controlMessageObserver, TimeoutsAndLimits, log), TimeoutsAndLimits, activeTcpConnectionsLimiter, log);
+            return (stream, log) => new MessageExchangeProtocol(new MessageExchangeStream(stream, messageSerializer, controlMessageObserver, TimeoutsAndLimits, log), TimeoutsAndLimits, activeTcpConnectionsLimiter, identityObserver, log);
         }
 
         public int Listen(IPEndPoint endpoint)
@@ -122,7 +125,8 @@ public int Listen(IPEndPoint endpoint)
                 HandleUnauthorizedClientConnect,
                 TimeoutsAndLimits,
                 streamFactory,
-                connectionsObserver);
+                connectionsObserver,
+                identityObserver);
 
             listeners.DoWithExclusiveAccess(l =>
             {
diff --git a/source/Halibut/HalibutRuntimeBuilder.cs b/source/Halibut/HalibutRuntimeBuilder.cs
@@ -27,6 +27,7 @@ public class HalibutRuntimeBuilder
         IRpcObserver? rpcObserver;
         IConnectionsObserver? connectionsObserver;
         IControlMessageObserver? controlMessageObserver;
+        IIdentityObserver? identityObserver;
 
         public HalibutRuntimeBuilder WithConnectionsObserver(IConnectionsObserver connectionsObserver)
         {
@@ -125,6 +126,12 @@ public HalibutRuntimeBuilder WithRpcObserver(IRpcObserver rpcObserver)
             return this;
         }
 
+        public HalibutRuntimeBuilder WithIdentityObserver(IIdentityObserver identityObserver)
+        {
+            this.identityObserver = identityObserver;
+            return this;
+        }
+
         public HalibutRuntime Build()
         {
             var halibutTimeoutsAndLimits = this.halibutTimeoutsAndLimits;
@@ -157,6 +164,7 @@ public HalibutRuntime Build()
             var connectionsObserver = this.connectionsObserver ?? NoOpConnectionsObserver.Instance;
             var rpcObserver = this.rpcObserver ?? new NoRpcObserver();
             var controlMessageObserver = this.controlMessageObserver ?? new NoOpControlMessageObserver();
+            var identityObserver = this.identityObserver ?? NoIdentityObserver.Instance;
 
             var halibutRuntime = new HalibutRuntime(
                 serviceFactory,
@@ -171,7 +179,8 @@ public HalibutRuntime Build()
                 streamFactory,
                 rpcObserver,
                 connectionsObserver,
-                controlMessageObserver);
+                controlMessageObserver,
+                identityObserver);
 
             if (onUnauthorizedClientConnect is not null)
             {
diff --git a/source/Halibut/Transport/Observability/IIdentityObserver.cs b/source/Halibut/Transport/Observability/IIdentityObserver.cs
@@ -0,0 +1,23 @@
+﻿// Copyright 2012-2013 Octopus Deploy Pty. Ltd.
+// 
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+// 
+//   http://www.apache.org/licenses/LICENSE-2.0
+// 
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+using Halibut.Transport.Protocol;
+
+namespace Halibut.Transport.Observability
+{
+    public interface IIdentityObserver
+    {
+        void IdentityEstablished(RemoteIdentity identity);
+    }
+}
diff --git a/source/Halibut/Transport/Observability/NoIdentityObserver.cs b/source/Halibut/Transport/Observability/NoIdentityObserver.cs
@@ -0,0 +1,29 @@
+// Copyright 2012-2013 Octopus Deploy Pty. Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+using System;
+using Halibut.Transport.Protocol;
+
+namespace Halibut.Transport.Observability
+{
+    public class NoIdentityObserver : IIdentityObserver
+    {
+        static NoIdentityObserver? singleInstance;
+        public static NoIdentityObserver Instance => singleInstance ??= new NoIdentityObserver();
+
+        public void IdentityEstablished(RemoteIdentity identity)
+        {
+        }
+    }
+}
diff --git a/source/Halibut/Transport/Protocol/MessageExchangeProtocol.cs b/source/Halibut/Transport/Protocol/MessageExchangeProtocol.cs
@@ -5,6 +5,7 @@
 using Halibut.Diagnostics;
 using Halibut.Exceptions;
 using Halibut.ServiceModel;
+using Halibut.Transport.Observability;
 
 namespace Halibut.Transport.Protocol
 {
@@ -20,15 +21,21 @@ public class MessageExchangeProtocol
         readonly IMessageExchangeStream stream;
         readonly HalibutTimeoutsAndLimits halibutTimeoutsAndLimits;
         readonly IActiveTcpConnectionsLimiter activeTcpConnectionsLimiter;
+        readonly IIdentityObserver identityObserver;
         readonly ILog log;
         bool identified;
         volatile bool acceptClientRequests = true;
 
-        public MessageExchangeProtocol(IMessageExchangeStream stream, HalibutTimeoutsAndLimits halibutTimeoutsAndLimits, IActiveTcpConnectionsLimiter activeTcpConnectionsLimiter, ILog log)
+        public MessageExchangeProtocol(IMessageExchangeStream stream,
+            HalibutTimeoutsAndLimits halibutTimeoutsAndLimits,
+            IActiveTcpConnectionsLimiter activeTcpConnectionsLimiter,
+            IIdentityObserver identityObserver,
+            ILog log)
         {
             this.stream = stream;
             this.halibutTimeoutsAndLimits = halibutTimeoutsAndLimits;
             this.activeTcpConnectionsLimiter = activeTcpConnectionsLimiter;
+            this.identityObserver = identityObserver;
             this.log = log;
         }
 
@@ -105,6 +112,7 @@ async Task ReceiveAndProcessRequestAsSubscriberAsync(IMessageExchangeStream stre
         public async Task ExchangeAsServerAsync(Func<RequestMessage, Task<ResponseMessage>> incomingRequestProcessor, Func<RemoteIdentity, IPendingRequestQueue> pendingRequests, CancellationToken cancellationToken)
         {
             var identity = await GetRemoteIdentityAsync(cancellationToken);
+            identityObserver.IdentityEstablished(identity);
 
             //We might need to limit the connection, so by default, we create an unlimited connection lease
             var limitedConnectionLease = activeTcpConnectionsLimiter.CreateUnlimitedLease();
diff --git a/source/Halibut/Transport/SecureListener.cs b/source/Halibut/Transport/SecureListener.cs
@@ -48,6 +48,7 @@ public class SecureListener : IAsyncDisposable
         readonly HalibutTimeoutsAndLimits halibutTimeoutsAndLimits;
         readonly IStreamFactory streamFactory;
         readonly IConnectionsObserver connectionsObserver;
+        readonly IIdentityObserver identityObserver;
         ILog log;
         TcpListener listener;
         Thread? backgroundThread;
@@ -67,7 +68,8 @@ public SecureListener(
             Func<string, string, UnauthorizedClientConnectResponse> unauthorizedClientConnect,
             HalibutTimeoutsAndLimits halibutTimeoutsAndLimits,
             IStreamFactory streamFactory,
-            IConnectionsObserver connectionsObserver)
+            IConnectionsObserver connectionsObserver,
+            IIdentityObserver identityObserver)
         {
             this.endPoint = endPoint;
             this.serverCertificate = serverCertificate;
@@ -81,6 +83,7 @@ public SecureListener(
             this.halibutTimeoutsAndLimits = halibutTimeoutsAndLimits;
             this.streamFactory = streamFactory;
             this.connectionsObserver = connectionsObserver;
+            this.identityObserver = identityObserver;
             this.cts = new CancellationTokenSource();
             this.cancellationToken = cts.Token;
 

Original file line number	Diff line number	Diff line change
`@@ -27,7 +27,7 @@ public void SetUp()`
`27`	`27`	`stream.SetRemoteIdentity(new RemoteIdentity(RemoteIdentityType.Server));`
`28`	`28`	`var limits = new HalibutTimeoutsAndLimitsForTestsBuilder().Build();`
`29`	`29`	`var activeConnectionsLimiter = new ActiveTcpConnectionsLimiter(limits);`
`30`		`- protocol = new MessageExchangeProtocol(stream, new HalibutTimeoutsAndLimitsForTestsBuilder().Build(), activeConnectionsLimiter, Substitute.For<ILog>());`
	`30`	`+ protocol = new MessageExchangeProtocol(stream, new HalibutTimeoutsAndLimitsForTestsBuilder().Build(), activeConnectionsLimiter, NoIdentityObserver.Instance, Substitute.For<ILog>());`
`31`	`31`	`}`
`32`	`32`
`33`	`33`	`// TODO - ASYNC ME UP! ExchangeAsClientAsync cancellation`
Original file line number	Diff line number	Diff line change
`@@ -67,7 +67,7 @@ public async Task SecureClientClearsPoolWhenAllConnectionsCorrupt()`
`67`	`67`	`var connection = Substitute.For<IConnection>();`
`68`	`68`	`var limits = new HalibutTimeoutsAndLimitsForTestsBuilder().Build();`
`69`	`69`	`var activeConnectionLimiter = new ActiveTcpConnectionsLimiter(limits);`
`70`		`- connection.Protocol.Returns(new MessageExchangeProtocol(stream, limits, activeConnectionLimiter, log));`
	`70`	`+ connection.Protocol.Returns(new MessageExchangeProtocol(stream, limits, activeConnectionLimiter, NoIdentityObserver.Instance, log));`
`71`	`71`
`72`	`72`	`await connectionManager.ReleaseConnectionAsync(endpoint, connection, CancellationToken.None);`
`73`	`73`	`}`
`@@ -96,7 +96,7 @@ static MessageExchangeProtocol GetProtocol(Stream stream, ILog logger)`
`96`	`96`	`{`
`97`	`97`	`var limits = new HalibutTimeoutsAndLimitsForTestsBuilder().Build();`
`98`	`98`	`var activeConnectionLimiter = new ActiveTcpConnectionsLimiter(limits);`
`99`		`- return new MessageExchangeProtocol(new MessageExchangeStream(stream, new MessageSerializerBuilder(new LogFactory()).Build(), new NoOpControlMessageObserver(), limits, logger), limits, activeConnectionLimiter, logger);`
	`99`	`+ return new MessageExchangeProtocol(new MessageExchangeStream(stream, new MessageSerializerBuilder(new LogFactory()).Build(), new NoOpControlMessageObserver(), limits, logger), limits, activeConnectionLimiter, NoIdentityObserver.Instance, logger);`
`100`	`100`	`}`
`101`	`101`	`}`
`102`	`102`	`}`