[camel] branch camel-2.x updated: CAMEL-13387 camel-twitter-direct-message doesn't filter by filterOld parameter https://issues.apache.org/jira/browse/CAMEL-13387

[da...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

davsclaus pushed a commit to branch camel-2.x
in repository https://gitbox.apache.org/repos/asf/camel.git


The following commit(s) were added to refs/heads/camel-2.x by this push:
     new bc49e6b  CAMEL-13387 camel-twitter-direct-message doesn't filter by filterOld parameter https://issues.apache.org/jira/browse/CAMEL-13387
bc49e6b is described below

commit bc49e6bf908a9d9826f9d4558dde835b87b13523
Author: Claudio Miranda <cl...@claudius.com.br>
AuthorDate: Mon Apr 1 16:41:35 2019 -0300

    CAMEL-13387 camel-twitter-direct-message doesn't filter by filterOld parameter
    https://issues.apache.org/jira/browse/CAMEL-13387
---
 .../DirectMessageConsumerHandler.java              | 57 +++++++++++++++++++---
 1 file changed, 49 insertions(+), 8 deletions(-)

diff --git a/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/directmessage/DirectMessageConsumerHandler.java b/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/directmessage/DirectMessageConsumerHandler.java
index d550864..59dd766 100644
--- a/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/directmessage/DirectMessageConsumerHandler.java
+++ b/components/camel-twitter/src/main/java/org/apache/camel/component/twitter/directmessage/DirectMessageConsumerHandler.java
@@ -16,6 +16,7 @@
  */
 package org.apache.camel.component.twitter.directmessage;
 
+import java.util.Iterator;
 import java.util.List;
 
 import org.apache.camel.Exchange;
@@ -23,6 +24,7 @@ import org.apache.camel.component.twitter.TwitterEndpoint;
 import org.apache.camel.component.twitter.consumer.AbstractTwitterConsumerHandler;
 import org.apache.camel.component.twitter.consumer.TwitterEventType;
 import twitter4j.DirectMessage;
+import twitter4j.DirectMessageList;
 import twitter4j.TwitterException;
 
 /**
@@ -36,19 +38,58 @@ public class DirectMessageConsumerHandler extends AbstractTwitterConsumerHandler
 
     @Override
     public List<Exchange> pollConsume() throws TwitterException {
-        List<DirectMessage> directMessages =  getTwitter().getDirectMessages(endpoint.getProperties().getCount());
-        for (int i = 0; i < directMessages.size(); i++) {
-            setLastIdIfGreater(directMessages.get(i).getId());
+        // the first call doesn't require any cursor as parameters
+        DirectMessageList directMessages = directMessages(null, null);
+        if (directMessages.size() > 0) {
+            // the DM response list is in reverse chronological order, so the last id is the first element.
+            setLastId(directMessages.get(0).getId());
         }
-
         return TwitterEventType.DIRECT_MESSAGE.createExchangeList(endpoint, directMessages);
     }
 
+    private DirectMessageList directMessages(String previousCursor, String cursor) throws TwitterException {
+        // https://developer.twitter.com/en/docs/direct-messages/sending-and-receiving/api-reference/list-events
+        // if there are more DM to retrieve the the next_cursor parameter is set.
+        // however next_cursor is always set in the response, so we must handle pagination correctly
+        // after the first call, there is always a second call to check if there is new DM and we compare
+        // the previous and current next_cursor, if not equals there is pagination.
+        DirectMessageList directMessages;
+        if (cursor != null) {
+            directMessages = getTwitter().getDirectMessages(endpoint.getProperties().getCount(), cursor);
+        } else {
+            directMessages = getTwitter().getDirectMessages(endpoint.getProperties().getCount());
+        }
+
+        String nextCursor = directMessages.getNextCursor();
+        // the condition will always be false for the first call.
+        if (nextCursor != null && nextCursor.equals(previousCursor) || directMessages.isEmpty()) {
+            directMessages.clear();
+            return directMessages;
+        }
+
+        if (nextCursor != null) {
+            DirectMessageList directMessages2 = directMessages(cursor, nextCursor);
+            directMessages.addAll(directMessages2);
+        }
+
+        // filter out messages retrieved before.
+        if (endpoint.getProperties().isFilterOld()) {
+            Iterator<DirectMessage> iterator = directMessages.iterator();
+            while (iterator.hasNext()) {
+                DirectMessage dm = iterator.next();
+                long id = dm.getId();
+                if (getLastId() > 1 && id <= getLastId()) {
+                    iterator.remove();
+                }
+            }
+        }
+
+        return directMessages;
+    }
+
     @Override
     public List<Exchange> directConsume() throws TwitterException {
-        return TwitterEventType.DIRECT_MESSAGE.createExchangeList(
-            endpoint,
-            getTwitter().getDirectMessages(endpoint.getProperties().getCount())
-        );
+        DirectMessageList directMessages = directMessages(null, null);
+        return TwitterEventType.DIRECT_MESSAGE.createExchangeList(endpoint, directMessages);
     }
 }