XCOMMONS-3451: Realtime editing doesn't support clustering

tmortagne · tmortagne · commit 78cf964b7e75 · 2025-10-15T14:40:44.000+02:00
diff --git a/xwiki-platform-core/xwiki-platform-netflux/pom.xml b/xwiki-platform-core/xwiki-platform-netflux/pom.xml
@@ -34,5 +34,6 @@
   <modules>
     <module>xwiki-platform-netflux-api</module>
     <module>xwiki-platform-netflux-rest</module>
+    <module>xwiki-platform-netflux-remote</module>
   </modules>
 </project>
diff --git a/xwiki-platform-core/xwiki-platform-netflux/xwiki-platform-netflux-api/src/main/java/org/xwiki/netflux/internal/DefaultEntityChannelStore.java b/xwiki-platform-core/xwiki-platform-netflux/xwiki-platform-netflux-api/src/main/java/org/xwiki/netflux/internal/DefaultEntityChannelStore.java
@@ -101,7 +101,7 @@ private boolean hasRawChannel(EntityChannel channel)
     {
         Channel rawChannel = this.channelStore.get(channel.getKey());
         if (rawChannel != null) {
-            channel.setUserCount(rawChannel.getConnectedUsers().size());
+            channel.setUserCount(rawChannel.getUsers().size());
             return true;
         } else {
             return false;
diff --git a/xwiki-platform-core/xwiki-platform-netflux/xwiki-platform-netflux-api/src/main/java/org/xwiki/netflux/internal/EntityChannelScriptAuthorBot.java b/xwiki-platform-core/xwiki-platform-netflux/xwiki-platform-netflux-api/src/main/java/org/xwiki/netflux/internal/EntityChannelScriptAuthorBot.java
@@ -30,6 +30,7 @@
 import org.xwiki.component.annotation.Component;
 import org.xwiki.netflux.EntityChannel;
 import org.xwiki.netflux.EntityChannelStore;
+import org.xwiki.netflux.internal.user.local.LocalUser;
 import org.xwiki.user.CurrentUserReference;
 import org.xwiki.user.UserReference;
 import org.xwiki.user.UserReferenceResolver;
diff --git a/xwiki-platform-core/xwiki-platform-netflux/xwiki-platform-netflux-api/src/test/java/org/xwiki/netflux/internal/DefaultEntityChannelStoreTest.java b/xwiki-platform-core/xwiki-platform-netflux/xwiki-platform-netflux-api/src/test/java/org/xwiki/netflux/internal/DefaultEntityChannelStoreTest.java
@@ -27,6 +27,7 @@
 import org.mockito.Mock;
 import org.xwiki.model.reference.WikiReference;
 import org.xwiki.netflux.EntityChannel;
+import org.xwiki.netflux.internal.user.local.LocalUser;
 import org.xwiki.test.junit5.mockito.ComponentTest;
 import org.xwiki.test.junit5.mockito.InjectMockComponents;
 import org.xwiki.test.junit5.mockito.MockComponent;
@@ -77,18 +78,14 @@ void createAndGetChannel()
         assertSame(entityChannel, this.entityChannelStore.createChannel(this.entityReference, path));
 
         // Add an user to the channel.
-        User me = new User(this.session, "mflorea");
+        User me = new LocalUser(this.session, "mflorea");
         channel.getUsers().put(me.getName(), me);
 
         // Get should return the existing channel.
         assertSame(entityChannel, this.entityChannelStore.getChannel(this.entityReference, path).get());
         assertSame(entityChannel, this.entityChannelStore.getChannel(channel.getKey()).get());
         assertEquals(1, entityChannel.getUserCount());
 
-        // Disconnect the user and check again the user count.
-        me.setConnected(false);
-        assertEquals(0, this.entityChannelStore.getChannel(this.entityReference, path).get().getUserCount());
-
         // Disconnect the raw channel and check the entity channel.
         when(this.channelStore.get(channel.getKey())).thenReturn(null);
         assertFalse(this.entityChannelStore.getChannel(this.entityReference, path).isPresent());
diff --git a/xwiki-platform-core/xwiki-platform-netflux/xwiki-platform-netflux-api/src/test/java/org/xwiki/netflux/internal/EntityChannelScriptAuthorBotTest.java b/xwiki-platform-core/xwiki-platform-netflux/xwiki-platform-netflux-api/src/test/java/org/xwiki/netflux/internal/EntityChannelScriptAuthorBotTest.java
@@ -28,6 +28,7 @@
 import org.xwiki.model.reference.DocumentReference;
 import org.xwiki.netflux.EntityChannel;
 import org.xwiki.netflux.EntityChannelStore;
+import org.xwiki.netflux.internal.user.local.LocalUser;
 import org.xwiki.test.junit5.mockito.ComponentTest;
 import org.xwiki.test.junit5.mockito.InjectMockComponents;
 import org.xwiki.test.junit5.mockito.MockComponent;
@@ -105,7 +106,7 @@ void onJoinChannel()
     @Test
     void onChannelMessage()
     {
-        User sender = mock(User.class);
+        LocalUser sender = mock(LocalUser.class);
         Session session = mock(Session.class);
         when(sender.getSession()).thenReturn(session);
 
diff --git a/xwiki-platform-core/xwiki-platform-netflux/xwiki-platform-netflux-remote/pom.xml b/xwiki-platform-core/xwiki-platform-netflux/xwiki-platform-netflux-remote/pom.xml
@@ -0,0 +1,64 @@
+<?xml version="1.0" encoding="UTF-8"?>
+
+<!--
+ * See the NOTICE file distributed with this work for additional
+ * information regarding copyright ownership.
+ *
+ * This is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU Lesser General Public License as
+ * published by the Free Software Foundation; either version 2.1 of
+ * the License, or (at your option) any later version.
+ *
+ * This software is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this software; if not, write to the Free
+ * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
+ * 02110-1301 USA, or see the FSF site: http://www.fsf.org.
+-->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+  <modelVersion>4.0.0</modelVersion>
+  <parent>
+    <groupId>org.xwiki.platform</groupId>
+    <artifactId>xwiki-platform-netflux</artifactId>
+    <version>17.9.0-SNAPSHOT</version>
+  </parent>
+  <artifactId>xwiki-platform-netflux-remote</artifactId>
+  <name>XWiki Platform - Netflux - Remote</name>
+  <description>Clustering related real-time communication.</description>
+  <packaging>jar</packaging>
+  <properties>
+    <!-- Name to display by the Extension Manager -->
+    <xwiki.extension.name>Netflux Remote</xwiki.extension.name>
+    <xwiki.jacoco.instructionRatio>1.00</xwiki.jacoco.instructionRatio>
+  </properties>
+  <dependencies>
+    <dependency>
+      <groupId>org.xwiki.platform</groupId>
+      <artifactId>xwiki-platform-netflux-api</artifactId>
+      <version>${platform.version}</version>
+    </dependency>
+
+    <!-- Test dependencies -->
+    <dependency>
+      <groupId>org.xwiki.commons</groupId>
+      <artifactId>xwiki-commons-tool-test-component</artifactId>
+      <version>${commons.version}</version>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>jakarta.servlet</groupId>
+      <artifactId>jakarta.servlet-api</artifactId>
+      <scope>test</scope>
+    </dependency>
+    <dependency>
+      <groupId>jakarta.websocket</groupId>
+      <artifactId>jakarta.websocket-api</artifactId>
+      <scope>test</scope>
+    </dependency>
+  </dependencies>
+</project>
diff --git a/xwiki-platform-core/xwiki-platform-netflux/xwiki-platform-netflux-remote/src/main/java/org/xwiki/netflux/internal/RemoteUser.java b/xwiki-platform-core/xwiki-platform-netflux/xwiki-platform-netflux-remote/src/main/java/org/xwiki/netflux/internal/RemoteUser.java
@@ -0,0 +1,53 @@
+/*
+ * See the NOTICE file distributed with this work for additional
+ * information regarding copyright ownership.
+ *
+ * This is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU Lesser General Public License as
+ * published by the Free Software Foundation; either version 2.1 of
+ * the License, or (at your option) any later version.
+ *
+ * This software is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this software; if not, write to the Free
+ * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
+ * 02110-1301 USA, or see the FSF site: http://www.fsf.org.
+ */
+package org.xwiki.netflux.internal;
+
+/**
+ * A user accessing the current instance.
+ * 
+ * @version $Id$
+ * @since 17.10.0RC1
+ */
+public class RemoteUser extends User
+{
+    private final String instance;
+
+    /**
+     * Creates a new user with the specified name, using the given WebSocket session.
+     * 
+     * @param instance the identifier of the instance on which the user is connected
+     * @param name the user name
+     */
+    public RemoteUser(String instance, String name)
+    {
+        super(name);
+
+        this.instance = instance;
+    }
+
+    /**
+     * @return the identifier of the instance on which the user is connected
+     */
+    public String getInstance()
+    {
+        return instance;
+    }
+
+}
diff --git a/xwiki-platform-core/xwiki-platform-netflux/xwiki-platform-netflux-remote/src/main/resources/META-INF/components.txt b/xwiki-platform-core/xwiki-platform-netflux/xwiki-platform-netflux-remote/src/main/resources/META-INF/components.txt
@@ -0,0 +1 @@
+org.xwiki.netflux.internal.RemoteUserHandler
diff --git a/xwiki-platform-core/xwiki-platform-observation/xwiki-platform-observation-remote/src/main/java/org/xwiki/observation/remote/RemoteObservationManager.java b/xwiki-platform-core/xwiki-platform-observation/xwiki-platform-observation-remote/src/main/java/org/xwiki/observation/remote/RemoteObservationManager.java
@@ -35,24 +35,41 @@
 public interface RemoteObservationManager
 {
     /**
-     * Send a event in the different network channels.
+     * @return the configured {@link NetworkAdapter}
+     * @since 17.10.0RC1
+     */
+    @Unstable
+    NetworkAdapter getNetworkAdapter();
+
+    /**
+     * Send an event in the different network channels.
      * <p>
      * This method is not supposed to be used directly for a new event unless the user specifically want to bypass or
      * emulate {@link org.xwiki.observation.ObservationManager}.
      *
-     * @param event the event
+     * @param localEvent the event to send
      */
-    void notify(LocalEventData event);
+    void notify(LocalEventData localEvent);
+
+    /**
+     * Send an event to the target channel members.
+     * 
+     * @param localEvent the event to send
+     * @param targets the members to send the event to
+     * @since 17.10.0RC1
+     */
+    @Unstable
+    void notify(LocalEventData localEvent, List<NetworkMember> targets);
 
     /**
-     * Inject a remote event in the local {@link org.xwiki.observation.ObservationManager}.
+     * Inject an remote event in the local {@link org.xwiki.observation.ObservationManager}.
      * <p>
      * This method is not supposed to be used directly for a new event unless the user specifically want to bypass or
      * emulate network.
      *
-     * @param event the event
+     * @param remoteEvent the event
      */
-    void notify(RemoteEventData event);
+    void notify(RemoteEventData remoteEvent);
 
     /**
      * Stop a running channel.
diff --git a/xwiki-platform-core/xwiki-platform-observation/xwiki-platform-observation-remote/src/main/java/org/xwiki/observation/remote/RemoteObservationManagerContext.java b/xwiki-platform-core/xwiki-platform-observation/xwiki-platform-observation-remote/src/main/java/org/xwiki/observation/remote/RemoteObservationManagerContext.java
@@ -32,9 +32,24 @@ public interface RemoteObservationManagerContext
 {
     /**
      * Set the remote state to true.
+     * 
+     * @deprecated use {@link #pushRemoteState(NetworkMember)} instead
      */
+    @Deprecated(since = "17.10.0RC1")
     void pushRemoteState();
 
+    /**
+     * Store in the context metadata associated to the current event.
+     * 
+     * @param sender the member who sent the event which triggered this thread, or it's not called in a remote event
+     *            thread
+     * @since 17.10.0RC1
+     */
+    default void pushRemoteState(NetworkMember sender)
+    {
+        pushRemoteState();
+    }
+
     /**
      * Set the remote state to false.
      */
@@ -44,4 +59,10 @@ public interface RemoteObservationManagerContext
      * @return indicate if the event in the current thread is a remote event
      */
     boolean isRemoteState();
+
+    /**
+     * @return the member who sent the event which triggered this thread, or it's not called in a remote event thread
+     * @since 17.10.0RC1
+     */
+    NetworkMember getSender();
 }
diff --git a/xwiki-platform-core/xwiki-platform-observation/xwiki-platform-observation-remote/src/main/java/org/xwiki/observation/remote/internal/DefaultRemoteObservationManager.java b/xwiki-platform-core/xwiki-platform-observation/xwiki-platform-observation-remote/src/main/java/org/xwiki/observation/remote/internal/DefaultRemoteObservationManager.java
@@ -20,6 +20,7 @@
 package org.xwiki.observation.remote.internal;
 
 import java.util.Collection;
+import java.util.List;
 
 import javax.inject.Inject;
 import javax.inject.Singleton;
@@ -38,6 +39,7 @@
 import org.xwiki.observation.remote.LocalEventData;
 import org.xwiki.observation.remote.NetworkAdapter;
 import org.xwiki.observation.remote.NetworkChannel;
+import org.xwiki.observation.remote.NetworkMember;
 import org.xwiki.observation.remote.RemoteEventData;
 import org.xwiki.observation.remote.RemoteEventException;
 import org.xwiki.observation.remote.RemoteObservationManager;
@@ -133,6 +135,12 @@ public void initialize() throws InitializationException
         }
     }
 
+    @Override
+    public NetworkAdapter getNetworkAdapter()
+    {
+        return this.networkAdapter;
+    }
+
     @Override
     public void notify(LocalEventData localEvent)
     {
@@ -158,6 +166,23 @@ public void notify(LocalEventData localEvent)
         }
     }
 
+    @Override
+    public void notify(LocalEventData localEvent, List<NetworkMember> targets)
+    {
+        if (this.remoteEventManagerContext.isRemoteState()) {
+            // the event is a remote event
+            return;
+        }
+
+        // Convert local->remote
+        RemoteEventData remoteEvent = this.eventConverterManager.createRemoteEventData(localEvent);
+
+        // if remote event data is not filled it means the message should not be sent to the network
+        if (remoteEvent != null) {
+            this.networkAdapter.send(remoteEvent);
+        }
+    }
+
     @Override
     public void notify(RemoteEventData remoteEvent)
     {
diff --git a/xwiki-platform-core/xwiki-platform-observation/xwiki-platform-observation-remote/src/main/java/org/xwiki/observation/remote/internal/DefaultRemoteObservationManagerContext.java b/xwiki-platform-core/xwiki-platform-observation/xwiki-platform-observation-remote/src/main/java/org/xwiki/observation/remote/internal/DefaultRemoteObservationManagerContext.java
@@ -25,6 +25,7 @@
 import org.xwiki.component.annotation.Component;
 import org.xwiki.context.Execution;
 import org.xwiki.context.ExecutionContext;
+import org.xwiki.observation.remote.NetworkMember;
 import org.xwiki.observation.remote.RemoteObservationManagerContext;
 
 /**
@@ -43,6 +44,11 @@ public class DefaultRemoteObservationManagerContext implements RemoteObservation
      */
     private static final String REMOTESTATE = "observation.remote.remotestate";
 
+    /**
+     * The name of the properties containing the channel member which sent the message.
+     */
+    private static final String SENDER = "observation.remote.sender";
+
     /**
      * Used to store remote observation manager context properties.
      */
@@ -57,12 +63,32 @@ public boolean isRemoteState()
         return context != null && context.getProperty(REMOTESTATE) == Boolean.TRUE;
     }
 
+    @Override
+    public NetworkMember getSender()
+    {
+        ExecutionContext context = this.execution.getContext();
+
+        if (context != null) {
+            return (NetworkMember) context.getProperty(SENDER);
+        }
+
+        return null;
+    }
+
     @Override
     public void pushRemoteState()
     {
         this.execution.getContext().setProperty(REMOTESTATE, Boolean.TRUE);
     }
 
+    @Override
+    public void pushRemoteState(NetworkMember sender)
+    {
+        pushRemoteState();
+
+        this.execution.getContext().setProperty(SENDER, sender);
+    }
+
     @Override
     public void popRemoteState()
     {

Original file line number	Diff line number	Diff line change
`@@ -101,7 +101,7 @@ private boolean hasRawChannel(EntityChannel channel)`
`101`	`101`	`{`
`102`	`102`	`Channel rawChannel = this.channelStore.get(channel.getKey());`
`103`	`103`	`if (rawChannel != null) {`
`104`		`- channel.setUserCount(rawChannel.getConnectedUsers().size());`
	`104`	`+ channel.setUserCount(rawChannel.getUsers().size());`
`105`	`105`	`return true;`
`106`	`106`	`} else {`
`107`	`107`	`return false;`
Original file line number	Diff line number	Diff line change
`@@ -0,0 +1 @@`
	`1`	`+org.xwiki.netflux.internal.RemoteUserHandler`