XCOMMONS-3451: Realtime editing doesn't support clustering

Author: tmortagne

xwiki-commons-core/xwiki-commons-netflux/src/main/java/org/xwiki/netflux/internal/Bot.java

@@ -57,7 +57,7 @@ default boolean onJoinChannel(Channel channel)
      * @param sender the user who sent the message
      * @param message the message that was sent
      */
-    default void onUserMessage(User sender, List<Object> message)
+    default void onUserMessage(LocalUser sender, List<Object> message)
     {
         // Do nothing by default
     }

xwiki-commons-core/xwiki-commons-netflux/src/main/java/org/xwiki/netflux/internal/Channel.java

@@ -22,7 +22,6 @@
 import java.util.Deque;
 import java.util.LinkedHashMap;
 import java.util.LinkedList;
-import java.util.List;
 import java.util.Map;
 
 import org.apache.commons.lang3.builder.EqualsBuilder;
@@ -88,14 +87,6 @@ public Map<String, Bot> getBots()
         return this.bots;
     }
 
-    /**
-     * @return the list of users that are currently connected to this channel
-     */
-    public List<User> getConnectedUsers()
-    {
-        return this.users.values().stream().filter(user -> user.getSession() != null && user.isConnected()).toList();
-    }
-
     /**
      * @return the channel messages
      */

xwiki-commons-core/xwiki-commons-netflux/src/main/java/org/xwiki/netflux/internal/ChannelStore.java

@@ -54,9 +54,22 @@ public class ChannelStore
      */
     public Channel create()
     {
-        Channel channel = new Channel(this.idGenerator.generateChannelId());
+        return create(this.idGenerator.generateChannelId());
+    }
+
+    /**
+     * Creates a new channel with a passed key.
+     * 
+     * @param channelKey the identifier of the new channel
+     * @return the new channel
+     * @since 17.10.0RC1
+     */
+    public Channel create(String channelKey)
+    {
+        Channel channel = new Channel(channelKey);
         askBotsToJoin(channel);
         this.channelByKey.put(channel.getKey(), channel);
+
         return channel;
     }
 
@@ -85,6 +98,25 @@ public Channel get(String key)
         return this.channelByKey.get(key);
     }
 
+    /**
+     * Access an existing channel by its key.
+     * 
+     * @param key the channel key
+     * @param create if true, create the channel when it does not exist
+     * @return the corresponding channel
+     * @since 17.10.0RC1
+     */
+    public Channel get(String key, boolean create)
+    {
+        Channel channel = get(key);
+
+        if (channel == null) {
+            channel = create(key);
+        }
+
+        return channel;
+    }
+
     /**
      * Remove a channel from memory.
      * 
@@ -106,8 +138,7 @@ public void prune()
         try {
             long currentTime = System.currentTimeMillis();
             for (Channel channel : this.channelByKey.values()) {
-                if (channel.getConnectedUsers().isEmpty()
-                    && (currentTime - channel.getCreationDate()) > (1000 * 60 * 60 * 2)) {
+                if (channel.getUsers().isEmpty() && (currentTime - channel.getCreationDate()) > (1000 * 60 * 60 * 2)) {
                     remove(channel);
                 }
             }

xwiki-commons-core/xwiki-commons-netflux/src/main/java/org/xwiki/netflux/internal/DefaultLocalUserFactory.java

@@ -0,0 +1,49 @@
+/*
+ * 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;
+
+import jakarta.inject.Singleton;
+import jakarta.websocket.Session;
+
+import org.xwiki.component.annotation.Component;
+
+/**
+ * Default implementation of {@link LocalUserFactory}.
+ * <p>
+ * Use the session id as user id.
+ * 
+ * @version $Id$
+ * @since 17.10.0RC1
+ */
+@Component
+@Singleton
+public class DefaultLocalUserFactory implements LocalUserFactory
+{
+    @Override
+    public LocalUser createLocalUser(Session session)
+    {
+        return new LocalUser(session, session.getId());
+    }
+
+    protected String getId(Session session)
+    {
+        return session.getId();
+    }
+}

xwiki-commons-core/xwiki-commons-netflux/src/main/java/org/xwiki/netflux/internal/HistoryKeeper.java

@@ -30,6 +30,8 @@
 
 import org.slf4j.Logger;
 import org.xwiki.component.annotation.Component;
+import org.xwiki.netflux.internal.event.NetfluxMessageUserEvent;
+import org.xwiki.observation.ObservationManager;
 
 /**
  * Holds the key of the history keeper fake user that is added to all Netflux channels.
@@ -51,6 +53,9 @@ public class HistoryKeeper extends AbstractBot
     @Inject
     private MessageBuilder messageBuilder;
 
+    @Inject
+    private ObservationManager observation;
+
     @Override
     public String getId()
     {
@@ -60,7 +65,7 @@ public String getId()
     }
 
     @Override
-    public void onUserMessage(User sender, List<Object> message)
+    public void onUserMessage(LocalUser sender, List<Object> message)
     {
         // The history keeper responds only to GET_HISTORY messages.
 
@@ -99,7 +104,7 @@ private void sendChannelHistory(User user, String channelKey)
 
         try {
             for (String msg : (Iterable<String>) messages::iterator) {
-                user.getSession().getBasicRemote().sendText(msg);
+                this.observation.notify(new NetfluxMessageUserEvent(user.getName(), msg), null);
             }
         } catch (Exception e) {
             this.logger.debug("Failed to send channel history.", e);

xwiki-commons-core/xwiki-commons-netflux/src/main/java/org/xwiki/netflux/internal/LocalUser.java

@@ -0,0 +1,54 @@
+/*
+ * 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;
+
+import jakarta.websocket.Session;
+
+/**
+ * A user accessing the current instance.
+ * 
+ * @version $Id$
+ * @since 17.10.0RC1
+ */
+public class LocalUser extends User
+{
+    private final Session session;
+
+    /**
+     * Creates a new user with the specified name, using the given WebSocket session.
+     * 
+     * @param session the WebSocket session used to communicate with the user
+     * @param name the identifier of the user
+     */
+    public LocalUser(Session session, String name)
+    {
+        super(name);
+
+        this.session = session;
+    }
+
+    /**
+     * @return the WebSocket session
+     */
+    public Session getSession()
+    {
+        return this.session;
+    }
+}

xwiki-commons-core/xwiki-commons-netflux/src/main/java/org/xwiki/netflux/internal/LocalUserFactory.java

@@ -0,0 +1,37 @@
+/*
+ * 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;
+
+import jakarta.websocket.Session;
+
+/**
+ * Component in charge of generating a new local user for a given session.
+ * 
+ * @version $Id$
+ * @since 17.10.0RC1
+ */
+public interface LocalUserFactory
+{
+    /**
+     * @param session the WebSocket session
+     * @return the new instance of {@link LocalUser}
+     */
+    LocalUser createLocalUser(Session session);
+}