generalize CacheInvalidator to ServerBrodcast to support methodName /…

… Argument

Signum.Engine.Extensions/Cache/AzureServiceBusCacheInvalidator.cs → Signum.Engine.Extensions/Cache/Broadcast/AzureServiceBusBroadcast.cs

@@ -10,10 +10,10 @@ namespace Signum.Engine.Cache;
 //Never Tested, works only in theory
 //https://github.com/briandunnington/SynchronizedCache/blob/master/SynchronizedCache.cs
 //https://github.com/Azure/azure-sdk-for-net/blob/main/sdk/servicebus/Azure.Messaging.ServiceBus/MigrationGuide.md
-public class AzureServiceBusCacheInvalidator : ICacheMultiServerInvalidator, IAsyncDisposable
+public class AzureServiceBusBroadcast : IServerBroadcast, IAsyncDisposable
 {
-    public event Action<string>? ReceiveInvalidation;
-
+    public event Action<string, string>? Receive;
+
     ServiceBusAdministrationClient adminClient;
     ServiceBusClient client;
 
@@ -25,26 +25,26 @@ public class AzureServiceBusCacheInvalidator : ICacheMultiServerInvalidator, IAs
 
     public DateTime StartTime;
 
-    public AzureServiceBusCacheInvalidator(string namespaceConnectionString, string topicName = "cache-invalidation")
+    public AzureServiceBusBroadcast(string namespaceConnectionString, string topicName = "cache-invalidation")
     {
         this.TopicName = topicName;
         this.SubscriptionName = Environment.MachineName + "-" + Schema.Current.ApplicationName;
 
         adminClient = new ServiceBusAdministrationClient(namespaceConnectionString);
         client = new ServiceBusClient(namespaceConnectionString);
         sender = client.CreateSender(this.TopicName);
-    }
-
-
-
-    public void SendInvalidation(string cleanName)
+    }
+
+
+    public void Send(string methodName, string argument)
     {
         sender.SendMessageAsync(new ServiceBusMessage(BinaryData.FromObjectAsJson(new AzureInvalidationMessage
         {
             CreationDate = DateTime.UtcNow,
             OriginMachineName = Environment.MachineName,
             OriginApplicationName = Schema.Current.ApplicationName,
-            CleanName = cleanName,
+            MethodName = methodName,
+            Argument = argument,
         }, EntityJsonContext.FullJsonSerializerOptions))).Wait();
     }
 
@@ -79,7 +79,7 @@ async Task StartMessageListener()
 
     private Task Processor_ProcessErrorAsync(ProcessErrorEventArgs arg)
     {
-        arg.Exception.LogException(ex => ex.ControllerName = nameof(AzureServiceBusCacheInvalidator));
+        arg.Exception.LogException(ex => ex.ControllerName = nameof(AzureServiceBusBroadcast));
         return Task.CompletedTask;
     }
 
@@ -96,19 +96,25 @@ private async Task Processor_ProcessMessageAsync(ProcessMessageEventArgs arg)
                message.OriginApplicationName == Schema.Current.ApplicationName)
                 return;
 
-            ReceiveInvalidation?.Invoke(message.CleanName);
+            Receive?.Invoke(message.MethodName, message.Argument);
 
             await arg.CompleteMessageAsync(arg.Message);
         }catch (Exception ex)
         {
-            ex.LogException(ex => ex.ControllerName = nameof(AzureServiceBusCacheInvalidator));
+            ex.LogException(ex => ex.ControllerName = nameof(AzureServiceBusBroadcast));
         }
     }
 
     public async ValueTask DisposeAsync()
     {
         await this.client.DisposeAsync();
-    }
+    }
+
+    public override string ToString()
+    {
+        return $"{nameof(AzureServiceBusBroadcast)}(TopicName = {TopicName}, SubscriptionName = {SubscriptionName})";
+    }
+
 }
 
 #pragma warning disable CS8618 // Non-nullable field must contain a non-null value when exiting constructor. Consider declaring as nullable.
@@ -117,6 +123,8 @@ public class AzureInvalidationMessage
     public DateTime CreationDate;
     public string OriginMachineName;
     public string OriginApplicationName;
-    public string CleanName;
+    public string MethodName;
+
+    public string Argument { get; internal set; }
 }
 #pragma warning restore CS8618 // Non-nullable field must contain a non-null value when exiting constructor. Consider declaring as nullable.

Signum.Engine.Extensions/Cache/Broadcast/PostgresBroadcast.cs

@@ -0,0 +1,68 @@
+using Npgsql;
+using System.Diagnostics;
+
+namespace Signum.Engine.Cache;
+
+public class PostgresBroadcast : IServerBroadcast
+{
+    public event Action<string, string>? Receive;
+
+    public void Send(string methodName, string argument)
+    {
+        Executor.ExecuteNonQuery($"NOTIFY table_changed, '{methodName}/{Process.GetCurrentProcess().Id}/{argument}'");
+    }
+
+    public void Start()
+    {
+        Task.Run(() =>
+        {
+            try
+            {
+                var conn = (NpgsqlConnection)Connector.Current.CreateConnection();
+                conn.Open();
+                conn.Notification += (o, e) =>
+                {
+                    try
+                    {
+                        var methodName = e.Payload.Before('/');
+                        var after = e.Payload.After("/");
+
+                        var pid = int.Parse(after.Before("/"));
+                        var arguments = after.After("/");
+
+                        if (Process.GetCurrentProcess().Id != pid)
+                            Receive?.Invoke(methodName, arguments);
+                    }
+                    catch (Exception ex)
+                    {
+                        ex.LogException(a => a.ControllerName = nameof(PostgresBroadcast));
+                    }
+                };
+
+                using (var cmd = new NpgsqlCommand("LISTEN table_changed", conn))
+                {
+                    cmd.ExecuteNonQuery();
+                }
+
+                while (true)
+                {
+                    conn.Wait();   // Thread will block here
+                }
+            }
+            catch (Exception e)
+            {
+                e.LogException(a =>
+                {
+                    a.ControllerName = nameof(PostgresBroadcast);
+                    a.ActionName = "Fatal";
+                });
+            }
+        });
+    }
+
+
+    public override string ToString()
+    {
+        return $"{nameof(PostgresBroadcast)}()";
+    }
+}

Signum.Engine.Extensions/Cache/Broadcast/SimpleHttpBroadcast.cs

@@ -0,0 +1,108 @@
+
+using Azure.Messaging.ServiceBus;
+using Azure.Messaging.ServiceBus.Administration;
+using Npgsql;
+using Signum.Engine.Json;
+using Signum.Entities.Cache;
+using Signum.Services;
+using System.Diagnostics;
+using System.Net.Http;
+using System.Net.Http.Json;
+
+namespace Signum.Engine.Cache;
+
+
+public class SimpleHttpBroadcast : IServerBroadcast
+{
+
+    HttpClient client = new HttpClient();
+    readonly string bordcastSecretHash;
+    readonly string[] broadcastUrls;
+
+    public SimpleHttpBroadcast(string broadcastSecret, string broadcastUrls)
+    {
+        this.bordcastSecretHash = Convert.ToBase64String(Security.EncodePassword(broadcastSecret));
+        this.broadcastUrls = broadcastUrls
+            .SplitNoEmpty(new char[] { ';', ',' } /*In theory ; and , are valid in a URL, but since we talk only domain names or IPs...*/)
+            .Select(a => a.Trim())
+            .Where(a => a.HasText())
+            .ToArray();
+    }
+
+    public event Action<string, string>? Receive;
+
+    public void Start()
+    {
+    }
+
+    //Called from Controller
+    public void InvalidateTable(InvalidateTableRequest request)
+    {
+        if (this.bordcastSecretHash != request.SecretHash)
+            throw new InvalidOperationException("invalidationSecret does not match");
+
+        if (request.OriginMachineName == Environment.MachineName ||
+            request.OriginApplicationName == Schema.Current.ApplicationName)
+            return;
+
+        Receive?.Invoke(request.MethodName, request.Argument);
+    }
+
+    public void Send(string methodName, string argument)
+    {
+        var request = new InvalidateTableRequest
+        {
+            MethodName = methodName,
+            Argument = argument,
+            SecretHash = this.bordcastSecretHash,
+            OriginMachineName = Environment.MachineName,
+            OriginApplicationName = Schema.Current.ApplicationName,
+        };
+
+        foreach (var url in broadcastUrls)
+        {
+            string? errorBody = null; 
+            try
+            {
+                var fullUrl = url.TrimEnd('/') + "/api/cache/invalidateTable";
+
+                var json = JsonContent.Create(request, options: EntityJsonContext.FullJsonSerializerOptions /*SignumServer.JsonSerializerOptions*/);
+
+                var response = client.PostAsync(fullUrl, json).Result;
+
+                if (!response.IsSuccessStatusCode)
+                {
+                    errorBody = response.Content.ReadAsStringAsync().Result;
+                }
+
+            }
+
+            catch (Exception e)
+            {
+                e.LogException(a =>
+                {
+                    a.ControllerName = nameof(SimpleHttpBroadcast);
+                    a.Data.Text = errorBody;
+                });
+            }
+        }
+    }
+
+    public override string ToString()
+    {
+        return $"{nameof(SimpleHttpBroadcast)}(Urls={broadcastUrls.ToString(", ")})";
+    }
+
+
+}
+
+#pragma warning disable CS8618 // Non-nullable field must contain a non-null value when exiting constructor. Consider declaring as nullable.
+public class InvalidateTableRequest
+{
+    public string OriginMachineName;
+    public string OriginApplicationName;
+    public string SecretHash;
+    public string Argument;
+    public string MethodName;
+}
+#pragma warning restore CS8618 // Non-nullable field must contain a non-null value when exiting constructor. Consider declaring as nullable.

Signum.Engine.Extensions/Cache/CacheLogic.cs

@@ -16,16 +16,16 @@
 
 namespace Signum.Engine.Cache;
 
-public interface ICacheMultiServerInvalidator
+public interface IServerBroadcast
 {
     void Start();
-    void SendInvalidation(string cleanName);
-    event Action<string>? ReceiveInvalidation;
+    void Send(string methodName, string argument);
+    event Action<string, string>? Receive;
 }
 
 public static class CacheLogic
 {
-    public static ICacheMultiServerInvalidator? CacheInvalidator { get; private set; }
+    public static IServerBroadcast? ServerBroadcast { get; private set; }
 
     public static bool WithSqlDependency { get; internal set; }
 
@@ -50,7 +50,7 @@ public static void AssertStarted(SchemaBuilder sb)
     ///    Change Server Authentication mode and enable SA: http://msdn.microsoft.com/en-us/library/ms188670.aspx
     ///    Change Database ownership to sa: ALTER AUTHORIZATION ON DATABASE::yourDatabase TO sa
     /// </summary>
-    public static void Start(SchemaBuilder sb, bool? withSqlDependency = null, ICacheMultiServerInvalidator? cacheInvalidator = null)
+    public static void Start(SchemaBuilder sb, bool? withSqlDependency = null, IServerBroadcast? serverBroadcast = null)
     {
         if (sb.NotDefined(MethodInfo.GetCurrentMethod()))
         {
@@ -63,14 +63,14 @@ public static void Start(SchemaBuilder sb, bool? withSqlDependency = null, ICach
 
             WithSqlDependency = withSqlDependency ?? Connector.Current.SupportsSqlDependency;
 
-            if (cacheInvalidator != null && WithSqlDependency)
+            if (serverBroadcast != null && WithSqlDependency)
                 throw new InvalidOperationException("cacheInvalidator is only necessary if SqlDependency is not enabled");
 
-            CacheInvalidator = cacheInvalidator;
-            if(CacheInvalidator != null)
+            ServerBroadcast = serverBroadcast;
+            if(ServerBroadcast != null)
             {
-                CacheInvalidator!.ReceiveInvalidation += CacheInvalidator_ReceiveInvalidation;
-                sb.Schema.BeforeDatabaseAccess += () => CacheInvalidator!.Start();
+                ServerBroadcast!.Receive += ServerBroadcast_Receive;
+                sb.Schema.BeforeDatabaseAccess += () => ServerBroadcast!.Start();
             }
 
             sb.Schema.SchemaCompleted += () => Schema_SchemaCompleted(sb);
@@ -115,7 +115,17 @@ static void Schema_SchemaCompleted(SchemaBuilder sb)
         }
     }
 
-    static void CacheInvalidator_ReceiveInvalidation(string cleanName)
+    static void ServerBroadcast_Receive(string methodName, string argument)
+    {
+        BroadcastReceivers.TryGetC(methodName)?.Invoke(argument);
+    }
+
+    public static Dictionary<string, Action<string>> BroadcastReceivers = new Dictionary<string, Action<string>>
+    {
+        { InvalidateTable, ServerBroadcast_InvalidateTable}
+    };
+
+    static void ServerBroadcast_InvalidateTable(string cleanName)
     {
         Type type = TypeEntity.TryGetType(cleanName)!;
 
@@ -707,6 +717,8 @@ This may be because SchemaCompleted is not yet called and you are accesing some
         }
     }
 
+    const string InvalidateTable = "InvalidateTable";
+
     internal static void NotifyInvalidateAllConnectedTypes(Type type)
     {
         var connected = inverseDependencies.IndirectlyRelatedTo(type, includeInitialNode: true);
@@ -717,7 +729,7 @@ internal static void NotifyInvalidateAllConnectedTypes(Type type)
             if (controller != null)
                 controller.NotifyInvalidated();
 
-            CacheInvalidator?.SendInvalidation(TypeLogic.GetCleanName(stype));
+            ServerBroadcast?.Send(InvalidateTable, TypeLogic.GetCleanName(stype));
         }
     }