camel-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From dhira...@apache.org
Subject git commit: Updated Box.com component to support long polling for updates from Box.com, fixed minor issues with token handling
Date Tue, 01 Jul 2014 16:29:49 GMT
Repository: camel
Updated Branches:
  refs/heads/master df062eaf4 -> 75fc0671b


Updated Box.com component to support long polling for updates from Box.com, fixed minor issues
with token handling


Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/75fc0671
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/75fc0671
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/75fc0671

Branch: refs/heads/master
Commit: 75fc0671bdc118d4d8e399889543431309ae63fe
Parents: df062ea
Author: Dhiraj Bokde <dhirajsb@yahoo.com>
Authored: Tue Jul 1 09:29:06 2014 -0700
Committer: Dhiraj Bokde <dhirajsb@yahoo.com>
Committed: Tue Jul 1 09:29:06 2014 -0700

----------------------------------------------------------------------
 components/camel-box/pom.xml                    |  91 +++++-
 .../apache/camel/component/box/BoxConsumer.java | 104 ++++++-
 .../apache/camel/component/box/BoxEndpoint.java |  22 +-
 .../apache/camel/component/box/BoxProducer.java |   4 -
 .../component/box/internal/BoxClientHelper.java |   8 +-
 .../component/box/internal/BoxConstants.java    |   2 +
 .../component/box/internal/EventCallback.java   |  29 ++
 .../box/internal/LongPollingEventsManager.java  | 302 +++++++++++++++++++
 .../box/internal/OAuthHelperListener.java       |   4 +
 .../signatures/long-polling-events-manager.txt  |   1 +
 .../component/box/AbstractBoxTestSupport.java   |   1 +
 .../box/IBoxFilesManagerIntegrationTest.java    |   1 -
 ...LongPollingEventsManagerIntegrationTest.java | 108 +++++++
 13 files changed, 660 insertions(+), 17 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/camel/blob/75fc0671/components/camel-box/pom.xml
----------------------------------------------------------------------
diff --git a/components/camel-box/pom.xml b/components/camel-box/pom.xml
index 26779de..b32661a 100644
--- a/components/camel-box/pom.xml
+++ b/components/camel-box/pom.xml
@@ -25,8 +25,6 @@
 
     <camel.osgi.export.pkg>${componentPackage}</camel.osgi.export.pkg>
     <camel.osgi.export.service>org.apache.camel.spi.ComponentResolver;component=box</camel.osgi.export.service>
-    <httpclient.version>4.3.3</httpclient.version>
-    <httpunit.version>1.7</httpunit.version>
   </properties>
 
   <dependencies>
@@ -35,6 +33,11 @@
       <artifactId>camel-core</artifactId>
     </dependency>
     <dependency>
+      <groupId>org.apache.httpcomponents</groupId>
+      <artifactId>httpclient</artifactId>
+      <version>${httpclient4-version}</version>
+    </dependency>
+    <dependency>
       <groupId>net.box</groupId>
       <artifactId>boxjavalibv2</artifactId>
       <version>${boxjavalibv2.version}</version>
@@ -47,6 +50,34 @@
           <groupId>org.apache.httpcomponents</groupId>
           <artifactId>httpclient-cache</artifactId>
         </exclusion>
+        <exclusion>
+          <groupId>com.fasterxml.jackson.core</groupId>
+          <artifactId>jackson-annotations</artifactId>
+        </exclusion>
+        <exclusion>
+          <groupId>com.fasterxml.jackson.core</groupId>
+          <artifactId>jackson-core</artifactId>
+        </exclusion>
+        <exclusion>
+          <groupId>com.fasterxml.jackson.core</groupId>
+          <artifactId>jackson-databind</artifactId>
+        </exclusion>
+        <exclusion>
+          <groupId>junit</groupId>
+          <artifactId>junit</artifactId>
+        </exclusion>
+        <exclusion>
+          <groupId>org.easymock</groupId>
+          <artifactId>easymock</artifactId>
+        </exclusion>
+        <exclusion>
+          <groupId>commons-io</groupId>
+          <artifactId>commons-io</artifactId>
+        </exclusion>
+        <exclusion>
+          <groupId>commons-lang</groupId>
+          <artifactId>commons-lang</artifactId>
+        </exclusion>
       </exclusions>
     </dependency>
     <dependency>
@@ -54,6 +85,11 @@
       <artifactId>htmlunit</artifactId>
       <version>${htmlunit.version}</version>
     </dependency>
+    <dependency>
+      <groupId>com.fasterxml.jackson.core</groupId>
+      <artifactId>jackson-databind</artifactId>
+      <version>${jackson2-version}</version>
+    </dependency>
 <!--
     <dependency>
       <groupId>httpunit</groupId>
@@ -63,8 +99,28 @@
 -->
     <dependency>
       <groupId>org.apache.httpcomponents</groupId>
+      <artifactId>httpcore</artifactId>
+      <version>${httpcore4-version}</version>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.httpcomponents</groupId>
+      <artifactId>httpmime</artifactId>
+      <version>${httpcore4-version}</version>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.httpcomponents</groupId>
       <artifactId>httpclient-cache</artifactId>
-      <version>${httpclient.version}</version>
+      <version>${httpclient4-version}</version>
+    </dependency>
+    <dependency>
+      <groupId>commons-io</groupId>
+      <artifactId>commons-io</artifactId>
+      <version>${commons-io-version}</version>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.commons</groupId>
+      <artifactId>commons-lang3</artifactId>
+      <version>${commons-lang3-version}</version>
     </dependency>
 
     <!-- Camel annotations in provided scope to avoid compile errors in IDEs -->
@@ -106,6 +162,26 @@
   <build>
     <plugins>
 
+      <!-- compile LongPollingEventsManager before generate-sources -->
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-compiler-plugin</artifactId>
+        <executions>
+          <execution>
+            <id>compile-long-polling-events-mnager</id>
+            <goals>
+              <goal>compile</goal>
+            </goals>
+            <phase>generate-sources</phase>
+            <configuration>
+              <includes>
+                <include>**/LongPollingEventsManager.java</include>
+              </includes>
+            </configuration>
+          </execution>
+        </executions>
+      </plugin>
+
       <!-- generate Component source and test source -->
       <plugin>
         <groupId>org.apache.camel</groupId>
@@ -229,6 +305,12 @@
                   </aliases>
                 </api>
                 <api>
+                  <apiName>poll-events</apiName>
+                  <proxyClass>org.apache.camel.component.box.internal.LongPollingEventsManager</proxyClass>
+                  <fromSignatureFile>${project.basedir}/src/signatures/long-polling-events-manager.txt</fromSignatureFile>
+                  <excludeConfigNames>callback</excludeConfigNames>
+                </api>
+                <api>
                   <apiName>search</apiName>
                   <proxyClass>com.box.boxjavalibv2.resourcemanagers.IBoxSearchManager</proxyClass>
                   <fromJavadoc/>
@@ -396,8 +478,9 @@
               <useFile>true</useFile>
               <forkMode>once</forkMode>
               <forkedProcessTimeoutInSeconds>300</forkedProcessTimeoutInSeconds>
+              <excludes/>
               <includes>
-                <include>**/*Test.java</include>
+                <include>**/*IntegrationTest.java</include>
               </includes>
             </configuration>
           </plugin>

http://git-wip-us.apache.org/repos/asf/camel/blob/75fc0671/components/camel-box/src/main/java/org/apache/camel/component/box/BoxConsumer.java
----------------------------------------------------------------------
diff --git a/components/camel-box/src/main/java/org/apache/camel/component/box/BoxConsumer.java
b/components/camel-box/src/main/java/org/apache/camel/component/box/BoxConsumer.java
index 3c2679f..f0da8a5 100644
--- a/components/camel-box/src/main/java/org/apache/camel/component/box/BoxConsumer.java
+++ b/components/camel-box/src/main/java/org/apache/camel/component/box/BoxConsumer.java
@@ -16,17 +16,115 @@
  */
 package org.apache.camel.component.box;
 
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+
+import com.box.boxjavalibv2.dao.BoxEventCollection;
+import com.box.boxjavalibv2.dao.BoxTypedObject;
+import org.apache.camel.Exchange;
 import org.apache.camel.Processor;
-import org.apache.camel.component.box.internal.BoxApiName;
-import org.apache.camel.util.component.AbstractApiConsumer;
+import org.apache.camel.component.box.internal.BoxConstants;
+import org.apache.camel.component.box.internal.CachedBoxClient;
+import org.apache.camel.component.box.internal.EventCallback;
+import org.apache.camel.component.box.internal.LongPollingEventsManager;
+import org.apache.camel.impl.DefaultConsumer;
+import org.apache.camel.util.ObjectHelper;
+import org.apache.camel.util.component.ApiConsumerHelper;
+import org.apache.camel.util.component.ApiMethod;
+import org.apache.camel.util.component.ApiMethodHelper;
+import org.apache.camel.util.component.PropertyNamesInterceptor;
+import org.apache.camel.util.component.ResultInterceptor;
 
 /**
  * The Box consumer.
  */
-public class BoxConsumer extends AbstractApiConsumer<BoxApiName, BoxConfiguration>
{
+//public class BoxConsumer extends AbstractApiConsumer<BoxApiName, BoxConfiguration>
{
+public class BoxConsumer extends DefaultConsumer
+    implements PropertyNamesInterceptor, ResultInterceptor, EventCallback {
+
+    private static final String CALLBACK_PROPERTY = "callback";
+
+    private final LongPollingEventsManager apiProxy;
+    private final Map<String, Object> properties;
+    private final ApiMethod apiMethod;
+
+    private boolean splitResult = true;
+    private CachedBoxClient cachedBoxClient;
 
     public BoxConsumer(BoxEndpoint endpoint, Processor processor) {
         super(endpoint, processor);
+
+        apiMethod = ApiConsumerHelper.findMethod(endpoint, this, log);
+
+        properties = new HashMap<String, Object>();
+        properties.putAll(endpoint.getEndpointProperties());
+        properties.put(CALLBACK_PROPERTY, this);
+
+        // invoke LongPollingEventsManager.poll to start event polling
+        cachedBoxClient = endpoint.getBoxClient();
+        apiProxy = new LongPollingEventsManager(cachedBoxClient,
+            endpoint.getConfiguration().getHttpParams(), endpoint.getExecutorService());
+    }
+
+    @Override
+    public void interceptPropertyNames(Set<String> propertyNames) {
+        propertyNames.add(CALLBACK_PROPERTY);
+    }
+
+    @Override
+    public void onEvent(BoxEventCollection events) throws Exception {
+        // convert Events to exchange and process
+        log.debug("Processed {} events for {}",
+            ApiConsumerHelper.getResultsProcessed(this, events, splitResult), cachedBoxClient);
+    }
+
+    @Override
+    public void onException(Exception e) {
+        getExceptionHandler().handleException(ObjectHelper.wrapRuntimeCamelException(e));
     }
 
+    @Override
+    protected void doStart() throws Exception {
+        super.doStart();
+
+        // invoke the API method to start polling
+        ApiMethodHelper.invokeMethod(apiProxy, apiMethod, properties);
+    }
+
+    @Override
+    protected void doStop() throws Exception {
+        super.doStop();
+        apiProxy.stopPolling();
+    }
+
+    public boolean isSplitResult() {
+        return splitResult;
+    }
+
+    public void setSplitResult(boolean splitResult) {
+        this.splitResult = splitResult;
+    }
+
+    @Override
+    public Object splitResult(Object result) {
+        if (result instanceof BoxEventCollection && splitResult) {
+            BoxEventCollection eventCollection = (BoxEventCollection) result;
+            final ArrayList<BoxTypedObject> entries = eventCollection.getEntries();
+            return entries.toArray(new BoxTypedObject[entries.size()]);
+        }
+        return result;
+    }
+
+    @Override
+    public void interceptResult(Object result, Exchange resultExchange) {
+        if (result instanceof BoxEventCollection) {
+            BoxEventCollection boxEventCollection = (BoxEventCollection) result;
+            resultExchange.getIn().setHeader(BoxConstants.CHUNK_SIZE_PROPERTY,
+                boxEventCollection.getChunkSize());
+            resultExchange.getIn().setHeader(BoxConstants.NEXT_STREAM_POSITION_PROPERTY,
+                boxEventCollection.getNextStreamPosition());
+        }
+    }
 }

http://git-wip-us.apache.org/repos/asf/camel/blob/75fc0671/components/camel-box/src/main/java/org/apache/camel/component/box/BoxEndpoint.java
----------------------------------------------------------------------
diff --git a/components/camel-box/src/main/java/org/apache/camel/component/box/BoxEndpoint.java
b/components/camel-box/src/main/java/org/apache/camel/component/box/BoxEndpoint.java
index d6f5ade..01c2678 100644
--- a/components/camel-box/src/main/java/org/apache/camel/component/box/BoxEndpoint.java
+++ b/components/camel-box/src/main/java/org/apache/camel/component/box/BoxEndpoint.java
@@ -26,6 +26,7 @@ import org.apache.camel.Producer;
 import org.apache.camel.component.box.internal.BoxApiCollection;
 import org.apache.camel.component.box.internal.BoxApiName;
 import org.apache.camel.component.box.internal.BoxClientHelper;
+import org.apache.camel.component.box.internal.BoxConstants;
 import org.apache.camel.component.box.internal.BoxPropertiesHelper;
 import org.apache.camel.component.box.internal.CachedBoxClient;
 import org.apache.camel.spi.UriEndpoint;
@@ -61,6 +62,11 @@ public class BoxEndpoint extends AbstractApiEndpoint<BoxApiName, BoxConfiguratio
     }
 
     public Producer createProducer() throws Exception {
+        // validate producer APIs
+        if (getApiName() == BoxApiName.POLL_EVENTS) {
+            throw new IllegalArgumentException("Producer endpoints do not support endpoint
prefix "
+                + BoxApiName.POLL_EVENTS.getName());
+        }
         return new BoxProducer(this);
     }
 
@@ -69,6 +75,12 @@ public class BoxEndpoint extends AbstractApiEndpoint<BoxApiName, BoxConfiguratio
         if (inBody != null) {
             throw new IllegalArgumentException("Option inBody is not supported for consumer
endpoint");
         }
+
+        // validate consumer APIs
+        if (getApiName() != BoxApiName.POLL_EVENTS) {
+            throw new IllegalArgumentException("Consumer endpoint only supports endpoint
prefix "
+                + BoxApiName.POLL_EVENTS.getName());
+        }
         final BoxConsumer consumer = new BoxConsumer(this, processor);
         // also set consumer.* properties
         configureConsumer(consumer);
@@ -80,6 +92,10 @@ public class BoxEndpoint extends AbstractApiEndpoint<BoxApiName, BoxConfiguratio
         return BoxPropertiesHelper.getHelper();
     }
 
+    protected String getThreadProfileName() {
+        return BoxConstants.THREAD_PROFILE_NAME;
+    }
+
     @Override
     protected void afterConfigureProperties() {
         // create client eagerly, a good way to validate configuration
@@ -120,7 +136,7 @@ public class BoxEndpoint extends AbstractApiEndpoint<BoxApiName, BoxConfiguratio
     }
 
     @Override
-    protected void interceptProperties(Map<String, Object> properties) {
+    public void interceptProperties(Map<String, Object> properties) {
         // set shared link and password from configuration if not set as header properties
         if (!properties.containsKey(SHARED_LINK_PROPERTY) && !ObjectHelper.isEmpty(sharedLink))
{
             properties.put(SHARED_LINK_PROPERTY, sharedLink);
@@ -227,4 +243,8 @@ public class BoxEndpoint extends AbstractApiEndpoint<BoxApiName, BoxConfiguratio
             super.doShutdown();
         }
     }
+
+    public CachedBoxClient getBoxClient() {
+        return cachedBoxClient;
+    }
 }

http://git-wip-us.apache.org/repos/asf/camel/blob/75fc0671/components/camel-box/src/main/java/org/apache/camel/component/box/BoxProducer.java
----------------------------------------------------------------------
diff --git a/components/camel-box/src/main/java/org/apache/camel/component/box/BoxProducer.java
b/components/camel-box/src/main/java/org/apache/camel/component/box/BoxProducer.java
index a1d2cf3..e08a59c 100644
--- a/components/camel-box/src/main/java/org/apache/camel/component/box/BoxProducer.java
+++ b/components/camel-box/src/main/java/org/apache/camel/component/box/BoxProducer.java
@@ -29,8 +29,4 @@ public class BoxProducer extends AbstractApiProducer<BoxApiName, BoxConfiguratio
     public BoxProducer(BoxEndpoint endpoint) {
         super(endpoint, BoxPropertiesHelper.getHelper());
     }
-
-    protected String getThreadProfileName() {
-        return BoxConstants.THREAD_PROFILE_NAME;
-    }
 }

http://git-wip-us.apache.org/repos/asf/camel/blob/75fc0671/components/camel-box/src/main/java/org/apache/camel/component/box/internal/BoxClientHelper.java
----------------------------------------------------------------------
diff --git a/components/camel-box/src/main/java/org/apache/camel/component/box/internal/BoxClientHelper.java
b/components/camel-box/src/main/java/org/apache/camel/component/box/internal/BoxClientHelper.java
index 621132f..6df4d1a 100644
--- a/components/camel-box/src/main/java/org/apache/camel/component/box/internal/BoxClientHelper.java
+++ b/components/camel-box/src/main/java/org/apache/camel/component/box/internal/BoxClientHelper.java
@@ -206,14 +206,14 @@ public final class BoxClientHelper {
 
                 LOG.debug("Revoking OAuth refresh token for {}", cachedBoxClient);
 
+                // revoke OAuth token
+                boxClient.getOAuthManager().revokeOAuth(boxClient.getAuthData().getAccessToken(),
+                    configuration.getClientId(), configuration.getClientSecret());
+
                 // notify the OAuthListener of revoked token
                 cachedBoxClient.getListener().onRefresh(null);
                 // mark auth data revoked
                 boxClient.getOAuthDataController().setOAuthData(null);
-
-                // revoke OAuth token
-                boxClient.getOAuthManager().revokeOAuth(boxClient.getAuthData().getAccessToken(),
-                    configuration.getClientId(), configuration.getClientSecret());
             }
         }
     }

http://git-wip-us.apache.org/repos/asf/camel/blob/75fc0671/components/camel-box/src/main/java/org/apache/camel/component/box/internal/BoxConstants.java
----------------------------------------------------------------------
diff --git a/components/camel-box/src/main/java/org/apache/camel/component/box/internal/BoxConstants.java
b/components/camel-box/src/main/java/org/apache/camel/component/box/internal/BoxConstants.java
index 11f584b..49a4474 100644
--- a/components/camel-box/src/main/java/org/apache/camel/component/box/internal/BoxConstants.java
+++ b/components/camel-box/src/main/java/org/apache/camel/component/box/internal/BoxConstants.java
@@ -23,6 +23,8 @@ public interface BoxConstants {
 
     // suffix for parameters when passed as exchange header properties
     String PROPERTY_PREFIX = "CamelBox.";
+    String NEXT_STREAM_POSITION_PROPERTY = PROPERTY_PREFIX + "nextStreamPosition";
+    String CHUNK_SIZE_PROPERTY = PROPERTY_PREFIX + "chunkSize";
 
     // thread profile name for this component
     String THREAD_PROFILE_NAME = "CamelBox";

http://git-wip-us.apache.org/repos/asf/camel/blob/75fc0671/components/camel-box/src/main/java/org/apache/camel/component/box/internal/EventCallback.java
----------------------------------------------------------------------
diff --git a/components/camel-box/src/main/java/org/apache/camel/component/box/internal/EventCallback.java
b/components/camel-box/src/main/java/org/apache/camel/component/box/internal/EventCallback.java
new file mode 100644
index 0000000..f217bf6
--- /dev/null
+++ b/components/camel-box/src/main/java/org/apache/camel/component/box/internal/EventCallback.java
@@ -0,0 +1,29 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.apache.camel.component.box.internal;
+
+import com.box.boxjavalibv2.dao.BoxEventCollection;
+
+/**
+ * Callback interface to handle BoxEvents received from long polling.
+ */
+public interface EventCallback {
+
+    void onEvent(BoxEventCollection events) throws Exception;
+
+    void onException(Exception e);
+}

http://git-wip-us.apache.org/repos/asf/camel/blob/75fc0671/components/camel-box/src/main/java/org/apache/camel/component/box/internal/LongPollingEventsManager.java
----------------------------------------------------------------------
diff --git a/components/camel-box/src/main/java/org/apache/camel/component/box/internal/LongPollingEventsManager.java
b/components/camel-box/src/main/java/org/apache/camel/component/box/internal/LongPollingEventsManager.java
new file mode 100644
index 0000000..66bb2b7
--- /dev/null
+++ b/components/camel-box/src/main/java/org/apache/camel/component/box/internal/LongPollingEventsManager.java
@@ -0,0 +1,302 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.apache.camel.component.box.internal;
+
+import java.net.SocketException;
+import java.net.SocketTimeoutException;
+import java.util.ArrayList;
+import java.util.Map;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+
+import com.box.boxjavalibv2.dao.BoxCollection;
+import com.box.boxjavalibv2.dao.BoxEventCollection;
+import com.box.boxjavalibv2.dao.BoxRealTimeServer;
+import com.box.boxjavalibv2.dao.BoxTypedObject;
+import com.box.boxjavalibv2.exceptions.AuthFatalFailureException;
+import com.box.boxjavalibv2.exceptions.BoxServerException;
+import com.box.boxjavalibv2.requests.requestobjects.BoxEventRequestObject;
+import com.box.boxjavalibv2.resourcemanagers.IBoxEventsManager;
+import com.box.restclientv2.exceptions.BoxRestException;
+import com.box.restclientv2.exceptions.BoxSDKException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.util.ObjectHelper;
+import org.apache.http.HttpEntity;
+import org.apache.http.HttpHeaders;
+import org.apache.http.HttpResponse;
+import org.apache.http.HttpStatus;
+import org.apache.http.StatusLine;
+import org.apache.http.client.HttpClient;
+import org.apache.http.client.methods.HttpGet;
+import org.apache.http.impl.client.DefaultHttpClient;
+import org.apache.http.params.BasicHttpParams;
+import org.apache.http.params.HttpConnectionParams;
+import org.apache.http.params.HttpParams;
+import org.apache.http.protocol.HttpContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Manager for monitoring events using long polling.
+ */
+@SuppressWarnings("deprecation")
+public class LongPollingEventsManager {
+
+    private static final Logger LOG = LoggerFactory.getLogger(LongPollingEventsManager.class);
+    private static final String RETRY_TIMEOUT = "retry_timeout";
+    private static final String MAX_RETRIES = "max_retries";
+    private static final String MESSAGE = "message";
+    private static final String NEW_CHANGE = "new_change";
+    private static final String RECONNECT = "reconnect";
+    private static final String OUT_OF_DATE = "out_of_date";
+
+    private final CachedBoxClient cachedBoxClient;
+    private final ExecutorService executorService;
+    private final BasicHttpParams httpParams;
+
+    private HttpClient httpClient;
+    private Future<?> pollFuture;
+    private HttpGet httpGet;
+    private boolean done;
+
+    public LongPollingEventsManager(CachedBoxClient boxClient,
+                                    Map<String, Object> httpParams, ExecutorService
executorService) {
+
+        this.cachedBoxClient = boxClient;
+        this.executorService = executorService;
+
+        this.httpParams = new BasicHttpParams();
+        HttpConnectionParams.setSoKeepalive(this.httpParams, true);
+
+        if (httpParams != null) {
+            for (Map.Entry<String, Object> entry : httpParams.entrySet()) {
+                this.httpParams.setParameter(entry.getKey(), entry.getValue());
+            }
+        }
+    }
+
+    @SuppressWarnings("unused")
+    public void poll(long streamPosition, final String streamType, final int limit, final
EventCallback callback)
+        throws BoxServerException, AuthFatalFailureException, BoxRestException {
+
+        // get BoxClient Event Manager
+        final IBoxEventsManager eventsManager = cachedBoxClient.getBoxClient().getEventsManager();
+
+        // get current stream position if requested
+        if (BoxEventRequestObject.STREAM_POSITION_NOW == streamPosition) {
+            streamPosition = getCurrentStreamPosition(eventsManager, streamPosition);
+        }
+
+        // validate parameters
+        ObjectHelper.notNull(streamPosition, "streamPosition");
+        ObjectHelper.notEmpty(streamType, "streamType");
+        ObjectHelper.notNull(callback, "eventCallback");
+
+        httpClient = new DefaultHttpClient(cachedBoxClient.getClientConnectionManager(),
httpParams);
+
+        // start polling thread
+        LOG.info("Started event polling thread for " + cachedBoxClient);
+
+        final long startStreamPosition = streamPosition;
+        pollFuture = executorService.submit(new Runnable() {
+            @Override
+            public void run() {
+
+                final ObjectMapper mapper = new ObjectMapper();
+
+                long currentStreamPosition = startStreamPosition;
+                BoxRealTimeServer realTimeServer = null;
+
+                boolean retry = false;
+                int retries = 0;
+                int maxRetries = 1;
+
+                while (!done) {
+                    try {
+                        // set to true if no exceptions thrown
+                        retry = false;
+
+                        if (realTimeServer == null) {
+
+                            // get RTS URL
+                            realTimeServer = getBoxRealTimeServer(currentStreamPosition,
eventsManager);
+
+                            // update HTTP timeout
+                            final int requestTimeout = Integer.parseInt(
+                                realTimeServer.getExtraData(RETRY_TIMEOUT).toString());
+                            final HttpParams params = httpClient.getParams();
+                            HttpConnectionParams.setSoTimeout(params, requestTimeout * 1000);
+
+                            // update maxRetries
+                            maxRetries = Integer.parseInt(realTimeServer.getExtraData(MAX_RETRIES).toString());
+                        }
+
+                        // create HTTP request for RTS
+                        httpGet = getPollRequest(realTimeServer.getUrl(), currentStreamPosition);
+
+                        // execute RTS poll
+                        HttpResponse httpResponse = null;
+                        try {
+                            httpResponse = httpClient.execute(httpGet, (HttpContext) null);
+                        } catch (SocketTimeoutException e) {
+                            LOG.debug("Poll timed out, retrying for " + cachedBoxClient);
+                        }
+
+                        if (httpResponse != null) {
+
+                            // parse response
+                            final StatusLine statusLine = httpResponse.getStatusLine();
+                            if (statusLine != null && statusLine.getStatusCode()
== HttpStatus.SC_OK) {
+                                final HttpEntity entity = httpResponse.getEntity();
+                                @SuppressWarnings("unchecked")
+                                Map<String, String> rtsResponse = mapper.readValue(entity.getContent(),
Map.class);
+
+                                final String message = rtsResponse.get(MESSAGE);
+                                if (NEW_CHANGE.equals(message)) {
+
+                                    // get events
+                                    final BoxEventRequestObject requestObject =
+                                        BoxEventRequestObject.getEventsRequestObject(currentStreamPosition);
+                                    requestObject.setStreamType(streamType);
+                                    requestObject.setLimit(limit);
+                                    final BoxEventCollection events = eventsManager.getEvents(requestObject);
+
+                                    // notify callback
+                                    callback.onEvent(events);
+
+                                    // update stream position
+                                    currentStreamPosition = events.getNextStreamPosition();
+
+                                } else if (RECONNECT.equals(message) || MAX_RETRIES.equals(message))
{
+                                    LOG.debug("Long poll reconnect for " + cachedBoxClient);
+                                    realTimeServer = null;
+                                } else if (OUT_OF_DATE.equals(message)) {
+                                    // update currentStreamPosition
+                                    LOG.debug("Long poll out of date for " + cachedBoxClient);
+                                    currentStreamPosition = getCurrentStreamPosition(eventsManager,
+                                        BoxEventRequestObject.STREAM_POSITION_NOW);
+                                    realTimeServer = null;
+                                } else {
+                                    throw new RuntimeCamelException("Unknown poll response
" + message);
+                                }
+                            } else {
+                                String msg = "Unknown error";
+                                if (statusLine != null) {
+                                    msg = String.format("Error polling events for %s: code=%s,
message=%s",
+                                        cachedBoxClient, statusLine.getStatusCode(), statusLine.getReasonPhrase());
+                                }
+                                throw new RuntimeCamelException(msg);
+                            }
+                        }
+
+                        // keep polling
+                        retry = true;
+
+                    } catch (InterruptedException e) {
+                        LOG.debug("Interrupted event polling thread for {}, exiting...",
cachedBoxClient);
+                    } catch (BoxSDKException e) {
+                        callback.onException(e);
+                    } catch (RuntimeCamelException e) {
+                        callback.onException(e);
+                    } catch (SocketException e) {
+                        // TODO handle connection aborts!!!
+                        LOG.debug("Socket exception while event polling for {}", cachedBoxClient);
+                        retry = true;
+                        realTimeServer = null;
+                    } catch (Exception e) {
+                        callback.onException(new RuntimeCamelException("Error while polling
for "
+                            + cachedBoxClient + ": " + e.getMessage(), e));
+                    } finally {
+                        // are we done yet?
+                        if (!retry) {
+                            done = true;
+                        } else {
+                            if (realTimeServer != null
+                                && (++retries > maxRetries)) {
+                                // make another option call
+                                realTimeServer = null;
+                            }
+                        }
+                    }
+                }
+                LOG.info("Stopped event polling thread for " + cachedBoxClient);
+            }
+        });
+    }
+
+    private long getCurrentStreamPosition(IBoxEventsManager eventsManager, long streamPosition)
+        throws BoxRestException, BoxServerException, AuthFatalFailureException {
+
+        final BoxEventRequestObject requestObject =
+            BoxEventRequestObject.getEventsRequestObject(streamPosition);
+        final BoxEventCollection events = eventsManager.getEvents(requestObject);
+        streamPosition = events.getNextStreamPosition();
+        return streamPosition;
+    }
+
+    public void stopPolling() throws Exception {
+        if (!done) {
+
+            // done polling
+            done = true;
+
+            // make sure an HTTP GET is not in progress
+            if (httpGet != null && !httpGet.isAborted()) {
+                httpGet.abort();
+            }
+
+            // cancel polling thread
+            if (pollFuture.cancel(true)) {
+                LOG.info("Stopped event polling for " + cachedBoxClient);
+            } else {
+                LOG.warn("Unable to stop event polling for " + cachedBoxClient);
+            }
+
+            httpClient = null;
+            pollFuture = null;
+        }
+    }
+
+    private BoxRealTimeServer getBoxRealTimeServer(long currentStreamPosition, IBoxEventsManager
eventsManager)
+        throws BoxRestException, BoxServerException, AuthFatalFailureException {
+
+        final BoxEventRequestObject optionsRequest =
+            BoxEventRequestObject.getEventsRequestObject(currentStreamPosition);
+
+        final BoxCollection eventOptions = eventsManager.getEventOptions(optionsRequest);
+        final ArrayList<BoxTypedObject> entries = eventOptions.getEntries();
+
+        // validate options
+        if (entries == null || entries.size() < 1
+            || !(entries.get(0) instanceof BoxRealTimeServer)) {
+
+            throw new RuntimeCamelException("No Real Time Server from event options for "
+ cachedBoxClient);
+        }
+
+        return (BoxRealTimeServer) entries.get(0);
+    }
+
+    private HttpGet getPollRequest(String url, long currentStreamPosition) throws AuthFatalFailureException
{
+
+        final HttpGet httpGet = new HttpGet(url + "&stream_position=" + currentStreamPosition);
+        final String accessToken = cachedBoxClient.getBoxClient().getAuthData().getAccessToken();
+        httpGet.setHeader(HttpHeaders.AUTHORIZATION, "Bearer " + accessToken);
+        return httpGet;
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/camel/blob/75fc0671/components/camel-box/src/main/java/org/apache/camel/component/box/internal/OAuthHelperListener.java
----------------------------------------------------------------------
diff --git a/components/camel-box/src/main/java/org/apache/camel/component/box/internal/OAuthHelperListener.java
b/components/camel-box/src/main/java/org/apache/camel/component/box/internal/OAuthHelperListener.java
index 684d1ba..7a62796 100644
--- a/components/camel-box/src/main/java/org/apache/camel/component/box/internal/OAuthHelperListener.java
+++ b/components/camel-box/src/main/java/org/apache/camel/component/box/internal/OAuthHelperListener.java
@@ -41,6 +41,7 @@ class OAuthHelperListener implements OAuthRefreshListener {
 
     @Override
     public void onRefresh(IAuthData newAuthData) {
+
         // look for refresh token update or revocation
         if (authSecureStorage != null
             && (newAuthData == null || !newAuthData.getRefreshToken().equals(refreshToken)))
{
@@ -49,5 +50,8 @@ class OAuthHelperListener implements OAuthRefreshListener {
         if (configListener != null) {
             configListener.onRefresh(newAuthData);
         }
+
+        // update cached refresh token
+        refreshToken = newAuthData != null ? newAuthData.getRefreshToken() : null;
     }
 }

http://git-wip-us.apache.org/repos/asf/camel/blob/75fc0671/components/camel-box/src/signatures/long-polling-events-manager.txt
----------------------------------------------------------------------
diff --git a/components/camel-box/src/signatures/long-polling-events-manager.txt b/components/camel-box/src/signatures/long-polling-events-manager.txt
new file mode 100644
index 0000000..57ea578
--- /dev/null
+++ b/components/camel-box/src/signatures/long-polling-events-manager.txt
@@ -0,0 +1 @@
+void poll(long streamPosition, final String streamType, final int limit, final org.apache.camel.component.box.internal.EventCallback
callback);
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/camel/blob/75fc0671/components/camel-box/src/test/java/org/apache/camel/component/box/AbstractBoxTestSupport.java
----------------------------------------------------------------------
diff --git a/components/camel-box/src/test/java/org/apache/camel/component/box/AbstractBoxTestSupport.java
b/components/camel-box/src/test/java/org/apache/camel/component/box/AbstractBoxTestSupport.java
index 7d2c33c..130bcaf 100644
--- a/components/camel-box/src/test/java/org/apache/camel/component/box/AbstractBoxTestSupport.java
+++ b/components/camel-box/src/test/java/org/apache/camel/component/box/AbstractBoxTestSupport.java
@@ -44,6 +44,7 @@ import org.junit.AfterClass;
 public abstract class AbstractBoxTestSupport extends CamelTestSupport {
 
     protected static final String CAMEL_TEST_TAG = "camel_was_here";
+    protected static final String CAMEL_TEST_FILE = "CamelTestFile";
     private static final String LINE_SEPARATOR = System.getProperty("line.separator");
     private static final String TEST_OPTIONS_PROPERTIES = "/test-options.properties";
     private static final String REFRESH_TOKEN_PROPERTY = "refreshToken";

http://git-wip-us.apache.org/repos/asf/camel/blob/75fc0671/components/camel-box/src/test/java/org/apache/camel/component/box/IBoxFilesManagerIntegrationTest.java
----------------------------------------------------------------------
diff --git a/components/camel-box/src/test/java/org/apache/camel/component/box/IBoxFilesManagerIntegrationTest.java
b/components/camel-box/src/test/java/org/apache/camel/component/box/IBoxFilesManagerIntegrationTest.java
index 03f483d..5fd8239 100644
--- a/components/camel-box/src/test/java/org/apache/camel/component/box/IBoxFilesManagerIntegrationTest.java
+++ b/components/camel-box/src/test/java/org/apache/camel/component/box/IBoxFilesManagerIntegrationTest.java
@@ -54,7 +54,6 @@ public class IBoxFilesManagerIntegrationTest extends AbstractBoxTestSupport
{
 
     private static final Logger LOG = LoggerFactory.getLogger(IBoxFilesManagerIntegrationTest.class);
     private static final String PATH_PREFIX = BoxApiName.FILES.getName();
-    private static final String CAMEL_TEST_FILE = "CamelTestFile";
     private static final BoxImageRequestObject BOX_IMAGE_REQUEST_OBJECT = BoxImageRequestObject.previewRequestObject();
     private static final String PNG_EXTENSION = "png";
     private static final String TEST_UPLOAD_FILE = "/log4j.properties";

http://git-wip-us.apache.org/repos/asf/camel/blob/75fc0671/components/camel-box/src/test/java/org/apache/camel/component/box/LongPollingEventsManagerIntegrationTest.java
----------------------------------------------------------------------
diff --git a/components/camel-box/src/test/java/org/apache/camel/component/box/LongPollingEventsManagerIntegrationTest.java
b/components/camel-box/src/test/java/org/apache/camel/component/box/LongPollingEventsManagerIntegrationTest.java
new file mode 100644
index 0000000..9502152
--- /dev/null
+++ b/components/camel-box/src/test/java/org/apache/camel/component/box/LongPollingEventsManagerIntegrationTest.java
@@ -0,0 +1,108 @@
+/*
+ * Camel Api Route test generated by camel-component-util-maven-plugin
+ * Generated on: Mon Jun 30 14:29:45 PDT 2014
+ */
+package org.apache.camel.component.box;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import com.box.boxjavalibv2.dao.BoxEvent;
+import com.box.boxjavalibv2.dao.BoxFile;
+import com.box.boxjavalibv2.requests.requestobjects.BoxItemCopyRequestObject;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.box.internal.BoxApiName;
+import org.apache.camel.component.box.internal.BoxConstants;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Test class for org.apache.camel.component.box.internal.LongPollingEventsManager APIs.
+ */
+public class LongPollingEventsManagerIntegrationTest extends AbstractBoxTestSupport {
+
+    private static final Logger LOG = LoggerFactory.getLogger(LongPollingEventsManagerIntegrationTest.class);
+
+    @Test
+    public void testPoll() throws Exception {
+
+        // generate file copy event
+        final Map<String, Object> headers = new HashMap<String, Object>();
+        headers.put("CamelBox.fileId", testFileId);
+        final BoxItemCopyRequestObject requestObject = BoxItemCopyRequestObject.copyItemRequestObject("0");
+        requestObject.setName(CAMEL_TEST_FILE);
+        headers.put("CamelBox.itemCopyRequest", requestObject);
+
+        BoxFile result = requestBodyAndHeaders("direct://COPYFILE", null, headers);
+
+        assertNotNull("copyFile result", result);
+        LOG.debug("copyFile: " + result);
+
+        // generate file delete event
+        headers.clear();
+        headers.put("CamelBox.fileId", result.getId());
+        headers.put("CamelBox.defaultRequest", null);
+        requestBodyAndHeaders("direct://DELETEFILE", null, headers);
+
+        MockEndpoint mockEndpoint = getMockEndpoint("mock:boxEvents");
+        mockEndpoint.expectedMinimumMessageCount(2);
+        mockEndpoint.setResultWaitTime(TimeUnit.MILLISECONDS.convert(30, TimeUnit.SECONDS));
+        mockEndpoint.assertIsSatisfied();
+
+        final List<Exchange> exchanges = mockEndpoint.getExchanges();
+        assertNotNull("poll result", exchanges);
+        assertFalse("poll result", exchanges.isEmpty());
+        LOG.debug("poll result: " + exchanges);
+
+        for (Exchange exchange : exchanges) {
+
+            assertNotNull("poll result " + BoxConstants.CHUNK_SIZE_PROPERTY,
+                exchange.getIn().getHeader(BoxConstants.CHUNK_SIZE_PROPERTY));
+            assertNotNull("poll result " + BoxConstants.NEXT_STREAM_POSITION_PROPERTY,
+                exchange.getIn().getHeader(BoxConstants.NEXT_STREAM_POSITION_PROPERTY));
+
+            final Object body = exchange.getIn().getBody();
+            assertNotNull("poll result body", body);
+            assertEquals("poll result body type", BoxEvent.class, body.getClass());
+
+            final String eventType = ((BoxEvent)body).getEventType();
+            assertTrue("poll result type",
+                BoxEvent.EVENT_TYPE_ITEM_COPY.equals(eventType) || BoxEvent.EVENT_TYPE_ITEM_TRASH.equals(eventType));
+        }
+    }
+
+    @Override
+    protected RouteBuilder createRouteBuilder() throws Exception {
+        return new RouteBuilder() {
+            public void configure() {
+                // test route for copyFile
+                from("direct://COPYFILE")
+                    .to("box://" + BoxApiName.FILES.getName() + "/copyFile");
+
+                // test route for deleteFile
+                from("direct://DELETEFILE")
+                    .to("box://" + BoxApiName.FILES.getName() + "/deleteFile");
+
+                // test route for poll
+                from("box://" + BoxApiName.POLL_EVENTS.getName() + "/poll?streamPosition=-1&streamType=all&limit=100")
+                    .to("mock:boxEvents");
+
+            }
+        };
+    }
+
+    @Override
+    protected CamelContext createCamelContext() throws Exception {
+        // also test revoke on shutdown
+        final CamelContext camelContext = super.createCamelContext();
+        final BoxComponent box = (BoxComponent) camelContext.getComponent("box");
+        box.getConfiguration().setRevokeOnShutdown(true);
+        return camelContext;
+    }
+}


Mime
View raw message