OF-1574: Add support for XEP-0352: Client State Indication

documentation/protocol-support.html

@@ -291,7 +291,7 @@ <h2>Mobile Compliance Support</h2>
 
         <table class="compliance">
             <tr>
-                <td>Feature</td>
+                <th>Feature</th>
                 <th>Specification</th>
                 <th>Core Server Supported</th>
                 <th>Advanced Server Supported</th>
@@ -303,8 +303,8 @@ <h2>Mobile Compliance Support</h2>
             </tr><tr>
                 <td>Client State Indication</td>
                 <td><a href="https://www.xmpp.org/extensions/xep-0352.html">XEP-0352</a>: Client State Indication</td>
-                <td class="unsupported"></td>
-                <td class="unsupported"></td>
+                <td class="supported"></td>
+                <td class="supported"></td>
             </tr><tr>
                 <td>Third Party Push Notifications</td>
                 <td><a href="https://www.xmpp.org/extensions/xep-0357.html">XEP-0357</a>: Push Notifications [<a href="#fn6">6</a>]</td>

i18n/src/main/resources/openfire_i18n.properties

@@ -1673,6 +1673,10 @@ system_property.sasl.scram-sha-1.iteration-count=The number of iterations when s
 system_property.xmpp.auth.anonymous=Set to true to allow anonymous login, otherwise false
 system_property.xmpp.auth.external.client.skip-cert-revalidation=Set to true to avoid validation of the client-provided PKIX certificate (for mutual authentication) other than the validation that happens when the TLS session is established.
 system_property.xmpp.auth.ssl.default-trustmanager-impl=The class to use as the default SSL/TLS TrustManager (which checks certificates from peers).
+system_property.xmpp.client.csi.enabled=Controls if Client State Indication (XEP-0352) functionality is supported by Openfire.
+system_property.xmpp.client.csi.delay.enabled=Determines if 'unimportant' stanzas are delayed for a client that is inactive.
+system_property.xmpp.client.csi.delay.max-duration=Determines the maximum duration of stanzas being delayed for a client that is inactive.
+system_property.xmpp.client.csi.delay.queue.capacity=Determines the maximum length of the queue that holds delayed stanzas.
 system_property.xmpp.client.idle=How long, in milliseconds, before idle client sessions are dropped. Set to -1 to never drop idle sessions.
 system_property.xmpp.client.idle.ping=Set to true to ping idle clients, otherwise false
 system_property.xmpp.client.version-query.enabled=Send a version request query to clients when they connect.
@@ -2106,6 +2110,10 @@ session.details.sm-detached=Detached
 session.details.sm-resume=Enabled (with resume)
 session.details.sm-enabled=Enabled (no resume)
 session.details.sm-disabled=Disabled
+session.details.csi-status=Client State Indication
+session.details.csi-active=Active
+session.details.csi-inactive=Inactive
+session.details.csi-delayed-stanzas=delayed stanzas
 session.details.connection-type=Connection Type
 session.details.software_version=Software Version
 session.details.cc-status=Message Carbons

i18n/src/main/resources/openfire_i18n_nl.properties

@@ -1592,6 +1592,10 @@ system_property.sasl.scram-sha-1.iteration-count=Het aantal iteraties dat gebrui
 system_property.xmpp.auth.anonymous=Zet op 'true' om anonieme aanmelding toe te staan.
 system_property.xmpp.auth.external.client.skip-cert-revalidation=Zet op 'true' om te voorkomen dat het PKIX certificaat dat door de client wordt aangeboden (voor mutual authentication) wordt gevalideerd, anders dan de validatie die plaats vindt wanneer de TLS sessie wordt opgezet.
 system_property.xmpp.auth.ssl.default-trustmanager-impl=De klasse die als standaard TLS TrustManager gebruikt wordt (om certificaten van andere netwerkentiteiten te controleren).
+system_property.xmpp.client.csi.enabled=Bepaalt of Client State Indication (XEP-0352) functionaliteit wordt ondersteund door Openfire.
+system_property.xmpp.client.csi.delay.enabled=Bepaalt of 'niet belangrijke' stanzas vertraagd worden afgeleverd aan de client, als deze inactief is.
+system_property.xmpp.client.csi.delay.max-duration=De maximale tijd dat stanzas worden vertraagd voor een client die inactief is.
+system_property.xmpp.client.csi.delay.queue.capacity=Het maximaal aantal voor een inactieve client vertraagde stanzas.
 system_property.xmpp.client.idle=Hoe lang, in milliseconden, een sessie die niet actief is wordt afgesloten. Zet op -1 om nooit inactieve sessies af te sluiten.
 system_property.xmpp.client.idle.ping=Zet op 'true' om inactieve sessies te pingen.
 system_property.xmpp.client.version-query.enabled=Probeer de versie van een client op te vragen, zodra er een nieuwe verbinding wordt gemaakt.
@@ -1600,8 +1604,8 @@ system_property.xmpp.gateway.enabled=Bepaald of Openfire's 'trunking' of 'gatewa
 system_property.xmpp.gateway.domains=Een collectie XMPP domeinnamen waar Openfire trunkingfunctionaliteit voor zal bieden. Openfire zal voor elk domein in deze collectie data accepteren van andere domeinen, om die data vervolgens door te sturen naar dat domein.
 system_property.xmpp.server.rewrite.replace-missing-to=Als de server een IQ of message stanza ontvangt zonder 'to' attribuut, zet dan de waarde van het 'to' attribuut op de bare JID representatie van het 'from' attribuut.
 system_property.xmpp.server.incoming.skip-jid-validation=Definieert of JIDs die de addressen zijn van stanzas die door externe domeinen worden aangeleverd, worden gevalideerd.
-system_property.xmpp.server.outgoing.max.threads=Minimaal aantal threads in de threadpool die gebruikt wordt om server-naar-server verbindingen op te zetten. 
-system_property.xmpp.server.outgoing.min.threads=Maximum aantal threads in de threadpool die gebruikt wordt om server-naar-server verbindingen op te zetten. 
+system_property.xmpp.server.outgoing.max.threads=Minimaal aantal threads in de threadpool die gebruikt wordt om server-naar-server verbindingen op te zetten.
+system_property.xmpp.server.outgoing.min.threads=Maximum aantal threads in de threadpool die gebruikt wordt om server-naar-server verbindingen op te zetten.
 system_property.xmpp.server.outgoing.threads-timeout=Tijd waarna inactieve en overbodige threads worden verwijderd van de threadpool die gebruikt wordt om server-naar-server verbindingen op te zetten.
 system_property.xmpp.server.outgoing.queue=Maximaal aantal uitgaande server-naar-server verbindingen die gelijktijdig met behulp van threads uit de threadpoool in het proces van verbinden zijn (overige connecties zullen worden gecreëerd door de aanroepende thread, waardoor andere operaties significant vertraagd kunnen worden).
 system_property.cluster-monitor.service-enabled=Zet op 'true' om voor cluster-events berichten naar administrators te sturen.
@@ -1996,6 +2000,10 @@ session.details.sm-detached=Ontkoppeld
 session.details.sm-resume=Ingeschakeld (met hervatten)
 session.details.sm-enabled=Ingeschakeld (zonder hervatten)
 session.details.sm-disabled=Uitgeschakeld
+session.details.csi-status=Client State Indication
+session.details.csi-active=Actief
+session.details.csi-inactive=Inactief
+session.details.csi-delayed-stanzas=vertraagde stanzas
 session.details.connection-type=Connectie Type
 session.details.software_version=Software Versie
 session.details.cc-status=Bericht Carbon Copy

xmppserver/src/main/java/org/jivesoftware/openfire/csi/CsiManager.java

@@ -0,0 +1,273 @@
+/*
+ * Copyright (C) 2023 Ignite Realtime Foundation. All rights reserved.
+ *
+ * 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.
+ */
+package org.jivesoftware.openfire.csi;
+
+import org.dom4j.Element;
+import org.jivesoftware.openfire.auth.UnauthorizedException;
+import org.jivesoftware.openfire.session.LocalClientSession;
+import org.jivesoftware.util.SystemProperty;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.xmpp.packet.IQ;
+import org.xmpp.packet.Message;
+import org.xmpp.packet.Packet;
+import org.xmpp.packet.Presence;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+import java.time.Duration;
+import java.time.Instant;
+import java.time.temporal.ChronoUnit;
+import java.util.Deque;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Set;
+
+/**
+ * Handles Client State Indication nonzas for one particular client session.
+ *
+ * @author Guus der Kinderen, [email protected]
+ * @see <a href="https://xmpp.org/extensions/xep-0352.html">XEP-0352: Client State Indication</a>
+ */
+public class CsiManager
+{
+    public static final Logger Log = LoggerFactory.getLogger(CsiManager.class);
+
+    /**
+     * Controls if Client State Indication functionality is made available to clients.
+     */
+    public static SystemProperty<Boolean> ENABLED = SystemProperty.Builder.ofType( Boolean.class )
+        .setKey("xmpp.client.csi.enabled")
+        .setDefaultValue(true)
+        .setDynamic(true)
+        .build();
+
+    /**
+     * Determines if 'unimportant' stanzas are delayed for a client that is inactive.
+     */
+    public static SystemProperty<Boolean> DELAY_ENABLED = SystemProperty.Builder.ofType( Boolean.class )
+        .setKey("xmpp.client.csi.delay.enabled")
+        .setDefaultValue(true)
+        .setDynamic(true)
+        .build();
+
+    /**
+     * Determines the maximum duration of stanzas being delayed for a client that is inactive.
+     */
+    public static SystemProperty<Duration> DELAY_MAX_DURATION = SystemProperty.Builder.ofType( Duration.class )
+        .setKey("xmpp.client.csi.delay.max-duration")
+        .setDefaultValue(Duration.ofMinutes(10))
+        .setChronoUnit(ChronoUnit.MILLIS)
+        .setDynamic(true)
+        .build();
+
+    /**
+     * Determines the maximum length of the queue that holds delayed stanzas.
+     */
+    public static SystemProperty<Integer> DELAY_QUEUE_CAPACITY = SystemProperty.Builder.ofType( Integer.class )
+        .setKey("xmpp.client.csi.delay.queue.capacity")
+        .setDefaultValue(500)
+        .setMinValue(0)
+        .setDynamic(true)
+        .build();
+
+    public static final String NAMESPACE = "urn:xmpp:csi:0";
+
+    /**
+     * The client session for which this instance manages CSI state.
+     */
+    private final LocalClientSession session;
+
+    /**
+     * Client state of {@link #session}, either 'true' for 'active', or 'false' for 'inactive'
+     */
+    private boolean active;
+
+    /**
+     * The timestamp of the last push of data to the client (or the time of instantiation of this instance, if no data
+     * has been pushed yet).
+     */
+    private Instant lastPush = Instant.now();
+
+    /**
+     * A queue that can hold 'unimportant' stanzas for the client if it is inactive.
+     */
+    private final Deque<Packet> queue = new LinkedList<>();
+
+    public CsiManager(@Nonnull final LocalClientSession session)
+    {
+        this.session = session;
+        this.active = true;
+    }
+
+    /**
+     * Processes a CSI nonza.
+     *
+     * @param nonza The CSI nonza to be processed.
+     */
+    public synchronized void process(@Nonnull final Element nonza)
+    {
+        switch(nonza.getName()) {
+            case "active":
+                activate();
+                break;
+            case "inactive":
+                deactivate();
+            default:
+                Log.warn("Unable to process element that was expected to be a CSI nonza for {}: {}", session, nonza);
+        }
+    }
+
+    /**
+     * Switch to the client state of 'active'.
+     */
+    public void activate()
+    {
+        Log.trace("Session for '{}' to CSI 'active'", session.getAddress());
+        active = true;
+
+        // If there are delayed stanzas, cause them to be delivered by rescheduling the last one.
+        if (!queue.isEmpty()) {
+            try {
+                session.deliver(queue.pollLast());
+            } catch (UnauthorizedException e) {
+                Log.error("Unexpected exception while activating CSI.", e);
+            }
+        }
+    }
+
+    /**
+     * Switch to the client state of 'inactive'.
+     */
+    public void deactivate()
+    {
+        Log.trace("Session for '{}' to CSI 'inactive'", session.getAddress());
+        active = false;
+    }
+
+    /**
+     * Returns the client state for the session that is being tracked by this instance, either 'true' for 'active',
+     * or 'false' for 'inactive'
+     *
+     * @return a client state indication
+     */
+    public boolean isActive()
+    {
+        return active;
+    }
+
+    /**
+     * Returns the number of stanzas that are currently in the delay queue.
+     *
+     * @return the number of delayed stanzas.
+     */
+    public synchronized int getDelayQueueSize() {
+        return queue.size();
+    }
+
+    /**
+     * Queues an unimportant stanza for later delivery, or returns the entire queue (including the argument) to be
+     * sent to the client.
+     *
+     * @param packet the stanza to process.
+     * @return stanzas to be delivered to the client (possibly empty).
+     */
+    public synchronized List<Packet> queueOrPush(@Nonnull final Packet packet)
+    {
+        queue.add(packet);
+
+        final boolean mustPush =
+               !DELAY_ENABLED.getValue() // The feature is disabled by configuration. Always send stanzas immediately.
+            || active // The client is active! Do not delay.
+            || queue.size() > DELAY_QUEUE_CAPACITY.getValue() // The delay queue has reached its capacity. Flush the entire thing.
+            || Instant.now().isAfter(lastPush.plus(DELAY_MAX_DURATION.getValue())) // Ensure that periodically, delayed data is sent anyway.
+            || !canDelay(packet);
+
+        final List<Packet> result = new LinkedList<>();
+        if (mustPush) {
+            result.addAll(queue);
+            queue.clear();
+            lastPush = Instant.now();
+            Log.trace("Cannot delay delivery of stanza. Push {} {}.", result.size(), result.size() == 1 ? "stanza" : "stanzas");
+        } else {
+            Log.trace("Delay delivery of stanza. Current queue size: {}", queue.size());
+        }
+        return result;
+    }
+
+    /**
+     * Inspects a stanza and evaluates if it is eligible for delayed delivery to inactive clients.
+     *
+     * @param stanza the stanza to inspect
+     * @return 'true' if the stanza delivery can be delayed.
+     */
+    static boolean canDelay(@Nonnull final Packet stanza)
+    {
+        if (stanza instanceof IQ) {
+            return false;
+        }
+
+        if (stanza instanceof Presence) {
+            final Presence presence = (Presence) stanza;
+            if (presence.getType() == null || presence.getType() == Presence.Type.unavailable) {
+                // Presence updates are generally unimportant, unless it is a MUC self-presence stanza, as that suggests
+                // that the user joined or left a room.
+                final Element muc = presence.getChildElement("x", "http://jabber.org/protocol/muc#user");
+                final boolean isSelfPresence = muc != null && muc.elements("status").stream().anyMatch(status -> "110".equals(status.attributeValue("code")));
+                return !isSelfPresence;
+            }
+        }
+
+        if (stanza instanceof Message)
+        {
+            final Message message = (Message) stanza;
+            if (message.getBody() == null) {
+                if (message.getType() == Message.Type.groupchat && !message.getElement().elements("subject").isEmpty()) {
+                    // A subject (which can be empty) is sent to indicate that a room join has completed.
+                    return false;
+                }
+
+                final Element muc = message.getChildElement("x", "http://jabber.org/protocol/muc#user");
+                if (muc != null && !muc.elements("invite").isEmpty()) {
+                    // Invitations to MUC rooms should be shown immediately.
+                    return false;
+                }
+
+                if (!message.getElement().elements("encrypted").isEmpty()) {
+                    // OMEMO messages never have a body element. We do not know what is being encrypted, but lets assume its important to err on the side of caution.
+                    return false;
+                }
+
+                // No message body, and none of the exemptions above? It can probably wait.
+                return true;
+            }
+        }
+
+        return false;
+    }
+
+    /**
+     * Checks if an XML fragment is recognized as a CSI nonza
+     *
+     * @param fragment the XML to evaluate
+     * @return true if the XML is recognized as a CSI nonza, otherwise false.
+     */
+    public static boolean isStreamManagementNonza(@Nullable final Element fragment) {
+        return fragment != null
+            && NAMESPACE.equals(fragment.getNamespaceURI())
+            && Set.of("active", "inactive").contains(fragment.getName());
+    }
+}

xmppserver/src/main/java/org/jivesoftware/openfire/csi/package-info.java

@@ -0,0 +1,8 @@
+/**
+ * Implementation of XEP-0352 "Client State Indication"
+ *
+ * It is common for IM clients to be logged in and 'online' even while the user is not interacting with the application. This protocol allows the client to indicate to the server when the user is not actively using the client, allowing the server to optimise traffic to the client accordingly. This can save bandwidth and resources on both the client and server.
+ *
+ * @see <a href="https://xmpp.org/extensions/xep-0352.html">XEP-0352: Client State Indication</a>
+ */
+package org.jivesoftware.openfire.csi;

xmppserver/src/main/java/org/jivesoftware/openfire/net/ClientStanzaHandler.java

@@ -20,8 +20,11 @@
 import org.jivesoftware.openfire.Connection;
 import org.jivesoftware.openfire.PacketRouter;
 import org.jivesoftware.openfire.auth.UnauthorizedException;
+import org.jivesoftware.openfire.csi.CsiManager;
 import org.jivesoftware.openfire.session.LocalClientSession;
 import org.jivesoftware.util.JiveGlobals;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.xmlpull.v1.XmlPullParser;
 import org.xmlpull.v1.XmlPullParserException;
 import org.xmpp.packet.IQ;
@@ -41,19 +44,19 @@
  */
 public class ClientStanzaHandler extends StanzaHandler {
 
+    private static final Logger Log = LoggerFactory.getLogger(ClientStanzaHandler.class);
+
     public ClientStanzaHandler(PacketRouter router, Connection connection) {
         super(router, connection);
     }
 
-    /**
-     * Only packets of type Message, Presence and IQ can be processed by this class. Any other
-     * type of packet is unknown and thus rejected generating the connection to be closed.
-     *
-     * @param doc the unknown DOM element that was received
-     * @return always false.
-     */
     @Override
     protected boolean processUnknowPacket(Element doc) {
+        if (CsiManager.isStreamManagementNonza(doc)) {
+            Log.trace("Client is sending client state indication nonza.");
+            ((LocalClientSession) session).getCsiManager().process(doc);
+            return true;
+        }
         return false;
     }