OF-1200: Monitoring plugin: Check if data has been flushed, cluster-wide.

src/plugins/monitoring/src/java/org/jivesoftware/openfire/archive/ConversationManager.java

@@ -39,6 +39,7 @@ import org.jivesoftware.openfire.XMPPServerInfo;
 import org.jivesoftware.openfire.archive.cluster.GetConversationCountTask;
 import org.jivesoftware.openfire.archive.cluster.GetConversationTask;
 import org.jivesoftware.openfire.archive.cluster.GetConversationsTask;
+import org.jivesoftware.openfire.archive.cluster.HasWrittenAllDataTask;
 import org.jivesoftware.openfire.cluster.ClusterManager;
 import org.jivesoftware.openfire.component.ComponentEventListener;
 import org.jivesoftware.openfire.component.InternalComponentManager;
@@ -988,11 +989,48 @@ public class ConversationManager implements Startable, ComponentEventListener{
      * content) to a request for archived data. Such response should only be generated after all data that was
      * queued before the request arrived has been written to the database.
      *
+     * This method performs a cluster-wide check, unlike {@link #hasLocalNodeWrittenAllDataBefore(Date)}.
+     *
      * @param date A date (cannot be null).
      * @return false if any of the the queues contain work that was created before the provided date, otherwise true.
      */
     public boolean hasWrittenAllDataBefore( Date date )
     {
+        final boolean localNode = hasLocalNodeWrittenAllDataBefore( date );
+        if ( !localNode )
+        {
+            return false;
+        }
+
+        // Check all other cluster nodes.
+        final Collection<Object> objects = CacheFactory.doSynchronousClusterTask( new HasWrittenAllDataTask( date ), false );
+        for ( final Object object : objects )
+        {
+            if ( !( (Boolean) object ) ) {
+                return false;
+            }
+        }
+        return true;
+    }
+
+    /**
+     * Returns true if none of the queues hold data that was delivered before the provided argument.
+     *
+     * This method is intended to be used to determine if it's safe to construct an answer (based on database
+     * content) to a request for archived data. Such response should only be generated after all data that was
+     * queued before the request arrived has been written to the database.
+     *
+     * This method performs a check on the local cluster-node only, unlike {@link #hasWrittenAllDataBefore(Date)}.
+     *
+     * @param date A date (cannot be null).
+     * @return false if any of the the queues contain work that was created before the provided date, otherwise true.
+     */
+    public boolean hasLocalNodeWrittenAllDataBefore( Date date )
+    {
+        if ( date == null )
+        {
+            throw new IllegalArgumentException( "Argument 'date' cannot be null." );
+        }
         final ArchiveCandidate c = conversationQueue.peek();
         final ArchiveCandidate m = messageQueue.peek();
         final ArchiveCandidate p = participantQueue.peek();

src/plugins/monitoring/src/java/org/jivesoftware/openfire/archive/cluster/HasWrittenAllDataTask.java

+/*
+ * Copyright (C) 2018 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.archive.cluster;
+
+import org.jivesoftware.openfire.XMPPServer;
+import org.jivesoftware.openfire.archive.ConversationManager;
+import org.jivesoftware.openfire.archive.MonitoringConstants;
+import org.jivesoftware.openfire.plugin.MonitoringPlugin;
+import org.jivesoftware.util.cache.ClusterTask;
+import org.jivesoftware.util.cache.ExternalizableUtil;
+
+import java.io.IOException;
+import java.io.ObjectInput;
+import java.io.ObjectOutput;
+import java.util.Date;
+
+/**
+ * A task that verifies if the data cache up to a specific point in time has been written to persistent storage.
+ *
+ * @author Guus der Kinderen, guus.der.kinderen@gmail.com
+ */
+public class HasWrittenAllDataTask implements ClusterTask<Boolean>
+{
+    private Date input;
+    private Boolean result;
+
+    public HasWrittenAllDataTask() {}
+
+    public HasWrittenAllDataTask( Date date )
+    {
+        this.input = date;
+    }
+
+    @Override
+    public void run()
+    {
+        final MonitoringPlugin plugin = (MonitoringPlugin) XMPPServer.getInstance().getPluginManager().getPlugin( MonitoringConstants.NAME );
+        final ConversationManager conversationManager = (ConversationManager) plugin.getModule( ConversationManager.class );
+        result = conversationManager.hasLocalNodeWrittenAllDataBefore( input );
+    }
+
+    @Override
+    public Boolean getResult()
+    {
+        return result;
+    }
+
+    @Override
+    public void writeExternal( ObjectOutput out ) throws IOException
+    {
+        ExternalizableUtil.getInstance().writeLong( out, input.getTime() );
+    }
+
+    @Override
+    public void readExternal( ObjectInput in ) throws IOException, ClassNotFoundException
+    {
+        input = new Date( ExternalizableUtil.getInstance().readLong( in ) );
+    }
+}