Permalink
Browse files

add queueing

  • Loading branch information...
1 parent 7d9403b commit c8eb1447700a4d2f794d626f79208f486227b312 @notnoop committed Jun 23, 2011
@@ -144,23 +144,27 @@ public MpnsServiceBuilder asQueued() {
*/
public MpnsService build() {
checkInitialization();
+ AbstractMpnsService service;
if (pooledMax == 1) {
HttpClient client = new DefaultHttpClient();
- return new MpnsServiceImpl(client);
+ service = new MpnsServiceImpl(client);
} else {
HttpClient client = new DefaultHttpClient(Utilities.poolManager(pooledMax));
- return new MpnsPooledService(client, executor);
+ service = new MpnsPooledService(client, executor);
}
+
+ if (isQueued) {
+ service = new MpnsQueuedService(service);
+ }
+
+ return service;
}
private void checkInitialization() {
if (pooledMax != 1 && executor == null) {
throw new IllegalStateException("Executor service is required for pooled connections");
}
- if (this.isQueued) {
- throw new RuntimeException("Queued Connections not supported yet");
- }
if (this.proxy != null) {
throw new RuntimeException("Proxies are not supported yet");
}
@@ -33,7 +33,6 @@
import java.util.Collection;
import java.util.Map.Entry;
-import org.apache.http.client.HttpClient;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.entity.ByteArrayEntity;
@@ -43,12 +42,6 @@
public abstract class AbstractMpnsService implements MpnsService {
- protected final HttpClient httpClient;
-
- protected AbstractMpnsService(HttpClient httpClient) {
- this.httpClient = httpClient;
- }
-
protected HttpPost postMessage(String subscriptionUri, byte[] requestBody,
Collection<? extends Entry<String, String>> headers) {
HttpPost method = new HttpPost(subscriptionUri);
@@ -77,7 +70,5 @@ public void push(String subscriptionUri, MpnsNotification message)
public void start() {}
- public void stop() {
- this.httpClient.getConnectionManager().shutdown();
- }
+ public void stop() {}
}
@@ -39,10 +39,11 @@
import com.notnoop.mpns.MpnsService;
public class MpnsPooledService extends AbstractMpnsService implements MpnsService {
+ private final HttpClient httpClient;
private final ExecutorService executor;
- public MpnsPooledService(HttpClient client, ExecutorService executor) {
- super(client);
+ public MpnsPooledService(HttpClient httpClient, ExecutorService executor) {
+ this.httpClient = httpClient;
this.executor = executor;
}
@@ -60,10 +61,10 @@ public void run() {
});
}
-
-
@Override
public void stop() {
+ super.stop();
+ this.httpClient.getConnectionManager().shutdown();
this.executor.shutdown();
}
@@ -0,0 +1,94 @@
+/*
+ * Copyright 2011, Mahmood Ali.
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ * * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ * * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following disclaimer
+ * in the documentation and/or other materials provided with the
+ * distribution.
+ * * Neither the name of Mahmood Ali. nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+package com.notnoop.mpns.internal;
+
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.apache.http.client.methods.HttpPost;
+
+import com.notnoop.mpns.MpnsService;
+
+public class MpnsQueuedService extends AbstractMpnsService implements MpnsService {
+
+ private AbstractMpnsService service;
+ private BlockingQueue<HttpPost> queue;
+ private AtomicBoolean started = new AtomicBoolean(false);
+
+ public MpnsQueuedService(AbstractMpnsService service) {
+ this.service = service;
+ this.queue = new LinkedBlockingQueue<HttpPost>();
+ }
+
+ @Override
+ protected void push(final HttpPost request) {
+ if (!started.get()) {
+ throw new IllegalStateException("Service hans't been started or was closed");
+ }
+
+ queue.add(request);
+ }
+
+ private Thread thread;
+ private volatile boolean shouldContinue;
+
+ public void start() {
+ if (started.getAndSet(true)) {
+ // Should we throw a runtime IllegalStateException here?
+ return;
+ }
+
+ service.start();
+ shouldContinue = true;
+ thread = new Thread() {
+ public void run() {
+ while (shouldContinue) {
+ try {
+ HttpPost post = queue.take();
+ service.push(post);
+ } catch (InterruptedException e) {}
+ }
+ }
+ };
+ thread.start();
+ }
+
+
+ @Override
+ public void stop() {
+ started.set(false);
+ shouldContinue = false;
+ thread.interrupt();
+ service.stop();
+ }
+
+}
@@ -41,8 +41,10 @@
import com.notnoop.mpns.exceptions.NetworkIOException;
public class MpnsServiceImpl extends AbstractMpnsService implements MpnsService {
- public MpnsServiceImpl(HttpClient client) {
- super(client);
+ protected final HttpClient httpClient;
+
+ public MpnsServiceImpl(HttpClient httpClient) {
+ this.httpClient = httpClient;
}
@Override
@@ -56,4 +58,9 @@ protected void push(HttpPost request) {
throw new NetworkIOException(e);
}
}
+
+ public void stop() {
+ this.httpClient.getConnectionManager().shutdown();
+ }
+
}

0 comments on commit c8eb144

Please sign in to comment.