Merge pull request igniterealtime#698 from Flowdalic/get-last-data-received

Add and use XMPPConnection.getLastDataReceived()

Author: Flowdalic

smack-core/src/main/java/org/jivesoftware/smack/AbstractXMPPConnection.java

@@ -2152,6 +2152,20 @@ public long getLastStanzaReceived() {
         return lastStanzaReceived;
     }
 
+    private long lastDataReceived;
+
+    protected void notifyDataReceived() {
+        lastDataReceived = System.currentTimeMillis();
+    }
+
+    @Override
+    public long getLastDataReceived() {
+        if (lastDataReceived == 0) {
+            return getLastStanzaReceived();
+        }
+        return lastDataReceived;
+    }
+
     @Override
     public JxmppContext getJxmppContext() {
         return jxmppContext;

smack-core/src/main/java/org/jivesoftware/smack/XMPPConnection.java

@@ -715,6 +715,13 @@ default boolean hasFeature(String element, String namespace) {
      */
     long getLastStanzaReceived();
 
+    /**
+     * Returns the timestamp in milliseconds when the last data was received.
+     *
+     * @return the timestamp in milliseconds
+     */
+    long getLastDataReceived();
+
     default JxmppContext getJxmppContext() {
         return JxmppContext.getDefaultContext();
     };

smack-core/src/main/java/org/jivesoftware/smack/c2s/ModularXmppClientToServerConnection.java

@@ -280,6 +280,10 @@ public void setTransport(XmppClientToServerTransport xmppTransport) {
                 ModularXmppClientToServerConnection.this.connected = true;
             }
 
+            @Override
+            public void notifyDataReceived() {
+                ModularXmppClientToServerConnection.this.notifyDataReceived();
+            }
         };
 
         // Construct the modules from the module descriptor. We do this before constructing the state graph, as the

smack-core/src/main/java/org/jivesoftware/smack/c2s/internal/ModularXmppClientToServerConnectionInternal.java

@@ -1,6 +1,6 @@
 /*
  *
- * Copyright 2020-2025 Florian Schmaus
+ * Copyright 2020-2026 Florian Schmaus
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -142,4 +142,7 @@ public abstract void waitForConditionOrThrowConnectionException(Supplier<Boolean
      * @param xmppTransport the active transport.
      */
     public abstract void setTransport(XmppClientToServerTransport xmppTransport);
+
+    public abstract void notifyDataReceived();
+
 }

smack-extensions/src/main/java/org/jivesoftware/smackx/ping/PingManager.java

@@ -1,6 +1,6 @@
 /*
  *
- * Copyright 2012-2018 Florian Schmaus
+ * Copyright 2012-2026 Florian Schmaus
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -416,11 +416,11 @@ public void pingServerIfNecessary() {
             // Ping has been disabled
             return;
         }
-        long lastStanzaReceived = connection.getLastStanzaReceived();
-        if (lastStanzaReceived > 0) {
+        long lastDataReceived = connection.getLastDataReceived();
+        if (lastDataReceived > 0) {
             long now = System.currentTimeMillis();
             // Delta since the last stanza was received
-            int deltaInSeconds = (int)  ((now - lastStanzaReceived) / 1000);
+            int deltaInSeconds = (int)  ((now - lastDataReceived) / 1000);
             // If the delta is small then the ping interval, then we can defer the ping
             if (deltaInSeconds < pingInterval) {
                 maybeSchedulePingServerTask(deltaInSeconds);

smack-tcp/src/main/java/org/jivesoftware/smack/tcp/CallbackInputStream.java

@@ -0,0 +1,58 @@
+/*
+ *
+ * Copyright 2026 Florian Schmaus
+ *
+ * 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.smack.tcp;
+
+import java.io.FilterInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.function.IntConsumer;
+
+class CallbackInputStream extends FilterInputStream {
+    private final IntConsumer onReadCallback;
+
+    CallbackInputStream(InputStream in, IntConsumer onReadCallback) {
+        super(in);
+        this.onReadCallback = onReadCallback != null ? onReadCallback : (bytes) -> { };
+    }
+
+    @Override
+    public int read() throws IOException {
+        int b = super.read();
+        if (b != -1) {
+            onReadCallback.accept(1);
+        }
+        return b;
+    }
+
+    @Override
+    public int read(byte[] b) throws IOException {
+        int bytesRead = super.read(b);
+        if (bytesRead > 0) {
+            onReadCallback.accept(bytesRead);
+        }
+        return bytesRead;
+    }
+
+    @Override
+    public int read(byte[] b, int off, int len) throws IOException {
+        int bytesRead = super.read(b, off, len);
+        if (bytesRead > 0) {
+            onReadCallback.accept(bytesRead);
+        }
+        return bytesRead;
+    }
+}

smack-tcp/src/main/java/org/jivesoftware/smack/tcp/XMPPTCPConnection.java

@@ -702,6 +702,7 @@ private void initConnection() throws IOException, InterruptedException {
 
     private void initReaderAndWriter() throws IOException {
         InputStream is = socket.getInputStream();
+        is = new CallbackInputStream(is, (bytesRead) -> notifyDataReceived());
         OutputStream os = socket.getOutputStream();
         if (compressionHandler != null) {
             is = compressionHandler.getInputStream(is);

smack-tcp/src/main/java/org/jivesoftware/smack/tcp/XmppTcpTransportModule.java

@@ -500,7 +500,10 @@ private void onChannelSelected(SelectableChannel selectedChannel, SelectionKey s
                     newInterestedOps |= SelectionKey.OP_WRITE;
                 }
 
-                callbackBytesRead += bytesRead;
+                if (bytesRead > 0) {
+                    callbackBytesRead += bytesRead;
+                    this.connectionInternal.notifyDataReceived();
+                }
 
                 ByteBuffer filteredIncomingBuffer = incomingBuffer;
                 for (ListIterator<XmppInputOutputFilter> it = connectionInternal.getXmppInputOutputFilterEndIterator(); it.hasPrevious();) {