Permalink
Browse files

Integrated new monitor for auto expiring requests in client

  • Loading branch information...
1 parent 0126bde commit 04a96223475cb939cdf97d219986ed74ad8fafa8 @jjlauer jjlauer committed May 3, 2011
View
2 pom.xml
@@ -100,7 +100,7 @@
<netty.version>[3.2,)</netty.version>
<ch-commons-charset.version>[2.0,)</ch-commons-charset.version>
<ch-commons-gsm.version>[2.0,)</ch-commons-gsm.version>
- <ch-commons-util.version>3.3-SNAPSHOT</ch-commons-util.version>
+ <ch-commons-util.version>4.2-SNAPSHOT</ch-commons-util.version>
</properties>
</project>
View
29 src/main/java/com/cloudhopper/smpp/SmppFuture.java
@@ -1,29 +0,0 @@
-/*
- * Copyright 2011 Twitter, Inc..
- *
- * 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 com.cloudhopper.smpp;
-
-import com.cloudhopper.commons.util.windowing.WindowFuture;
-import com.cloudhopper.smpp.pdu.PduRequest;
-import com.cloudhopper.smpp.pdu.PduResponse;
-
-/**
- * Interface representing either an asynchronous or synchronous operation in SMPP.
- *
- * @author joelauer
- */
-public interface SmppFuture extends WindowFuture<Integer,PduRequest,PduResponse> {
-
-}
View
8 src/main/java/com/cloudhopper/smpp/SmppSession.java
@@ -164,13 +164,19 @@
public long getBoundTime();
/**
+ * @deprecated
+ * @see #getRequestWindow
+ */
+ public Window<Integer,PduRequest,PduResponse> getRequestWindow();
+
+ /**
* Gets the underlying request "window" for this session. A "window" represents
* a request sent to the remote endpoint, but has not received a response
* yet. Accessing this property is useful if unacknowledged requests need
* to be cleared out (most likely for a retry at a later time).
* @return The request "window"
*/
- public Window<Integer,PduRequest,PduResponse> getRequestWindow();
+ public Window<Integer,PduRequest,PduResponse> getSendWindow();
/**
* Immediately close the session by closing the underlying socket/channel.
View
9 src/main/java/com/cloudhopper/smpp/impl/DefaultSmppClient.java
@@ -40,6 +40,7 @@
import com.cloudhopper.smpp.type.UnrecoverablePduException;
import java.net.InetSocketAddress;
import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ScheduledExecutorService;
import org.jboss.netty.bootstrap.ClientBootstrap;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelFuture;
@@ -62,6 +63,7 @@
private ExecutorService executors;
private NioClientSocketChannelFactory channelFactory;
private ClientBootstrap clientBootstrap;
+ private ScheduledExecutorService monitorExecutor;
public DefaultSmppClient() {
this(DaemonExecutors.newCachedDaemonThreadPool());
@@ -72,13 +74,18 @@ public DefaultSmppClient(ExecutorService executors) {
}
public DefaultSmppClient(ExecutorService executors, int expectedSessions) {
+ this(executors, expectedSessions, null);
+ }
+
+ public DefaultSmppClient(ExecutorService executors, int expectedSessions, ScheduledExecutorService monitorExecutor) {
this.channels = new DefaultChannelGroup();
this.executors = executors;
this.channelFactory = new NioClientSocketChannelFactory(this.executors, this.executors, expectedSessions);
this.clientBootstrap = new ClientBootstrap(channelFactory);
// we use the same default pipeline for all new channels - no need for a factory
this.clientConnector = new SmppClientConnector(this.channels);
this.clientBootstrap.getPipeline().addLast(SmppChannelConstants.PIPELINE_CLIENT_CONNECTOR_NAME, this.clientConnector);
+ this.monitorExecutor = monitorExecutor;
}
@Override
@@ -177,7 +184,7 @@ protected DefaultSmppSession doOpen(SmppSessionConfiguration config, SmppSession
}
protected DefaultSmppSession createSession(Channel channel, SmppSessionConfiguration config, SmppSessionHandler sessionHandler) throws SmppTimeoutException, SmppChannelException, InterruptedException {
- DefaultSmppSession session = new DefaultSmppSession(SmppSession.Type.CLIENT, config, channel, sessionHandler);
+ DefaultSmppSession session = new DefaultSmppSession(SmppSession.Type.CLIENT, config, channel, sessionHandler, monitorExecutor);
// add the thread renamer portion to the pipeline
if (config.getName() != null) {
View
14 src/main/java/com/cloudhopper/smpp/impl/DefaultSmppSession.java
@@ -21,7 +21,6 @@
import com.cloudhopper.commons.util.windowing.WindowListener;
import com.cloudhopper.smpp.SmppBindType;
import com.cloudhopper.smpp.SmppConstants;
-import com.cloudhopper.smpp.SmppFuture;
import com.cloudhopper.smpp.SmppServerSession;
import com.cloudhopper.smpp.type.SmppChannelException;
import com.cloudhopper.smpp.SmppSessionConfiguration;
@@ -255,9 +254,14 @@ public SequenceNumber getSequenceNumber() {
protected PduTranscoder getTranscoder() {
return this.transcoder;
}
-
+
@Override
public Window<Integer,PduRequest,PduResponse> getRequestWindow() {
+ return getSendWindow();
+ }
+
+ @Override
+ public Window<Integer,PduRequest,PduResponse> getSendWindow() {
return this.sendWindow;
}
@@ -410,8 +414,9 @@ protected PduResponse sendRequestAndGetResponse(PduRequest requestPdu, long time
boolean completedWithinTimeout = future.await();
if (!completedWithinTimeout) {
- // FIXME: make sure we remove this request from the window??
-// future.cancel();
+ // since this is a "synchronous" request and it timed out, we don't
+ // want it eating up valuable window space - cancel it before returning exception
+ future.cancel();
throw new SmppTimeoutException("Unable to get response within [" + timeoutInMillis + " ms]");
}
@@ -471,6 +476,7 @@ protected PduResponse sendRequestAndGetResponse(PduRequest requestPdu, long time
try {
future = sendWindow.offer(pdu.getSequenceNumber(), pdu, timeoutInMillis, configuration.getRequestExpiryTimeout(), synchronous);
logger.debug("IsCallerWaiting? " + future.isCallerWaiting());
+ logger.debug("Expire Timestamp: " + future.getExpireTimestamp());
} catch (DuplicateKeyException e) {
throw new UnrecoverablePduException(e.getMessage(), e);
} catch (OfferTimeoutException e) {
View
9 src/test/java/com/cloudhopper/smpp/demo/ClientMain.java
@@ -22,7 +22,6 @@
import com.cloudhopper.smpp.impl.DefaultSmppClient;
import com.cloudhopper.smpp.impl.DefaultSmppSessionHandler;
import com.cloudhopper.smpp.pdu.DeliverSm;
-import com.cloudhopper.smpp.pdu.SubmitSmResp;
import com.cloudhopper.smpp.type.Address;
import com.cloudhopper.smpp.pdu.EnquireLink;
import com.cloudhopper.smpp.pdu.PduRequest;
@@ -44,7 +43,9 @@ static public void main(String[] args) throws Exception {
// THIS VERSION USES "DAEMON" threads by default
// SmppSessionBootstrap bootstrap = new SmppSessionBootstrap();
// THIS VERSION DOESN'T - WILL HANG JVM UNTIL CLOSED
- DefaultSmppClient bootstrap = new DefaultSmppClient(Executors.newCachedThreadPool());
+ //DefaultSmppClient bootstrap = new DefaultSmppClient(Executors.newCachedThreadPool());
+ // include monitoring (automatic expiration) in this version
+ DefaultSmppClient bootstrap = new DefaultSmppClient(Executors.newCachedThreadPool(), 1, Executors.newScheduledThreadPool(1));
DefaultSmppSessionHandler sessionHandler = new ClientSmppSessionHandler();
@@ -58,6 +59,10 @@ static public void main(String[] args) throws Exception {
config0.setSystemId("1234567890");
config0.setPassword("password");
config0.getLoggingOptions().setLogBytes(true);
+
+ // to enable monitoring
+ config0.setRequestExpiryTimeout(2000);
+ config0.setWindowMonitorInterval(1000);
SmppSession session0 = null;
View
56 src/test/java/com/cloudhopper/smpp/impl/DefaultSmppSessionTest.java
@@ -342,7 +342,7 @@ public void windowSizeBlocksAsyncRequest() throws Exception {
WindowFuture future1 = session.sendRequestPdu(el1, 3000, true);
WindowFuture future2 = session.sendRequestPdu(el2, 3000, true);
- Assert.assertEquals(3, session.getRequestWindow().getSize());
+ Assert.assertEquals(3, session.getSendWindow().getSize());
try {
// window size of 3 is now filled up, this one should timeout
@@ -353,7 +353,7 @@ public void windowSizeBlocksAsyncRequest() throws Exception {
Assert.assertEquals(OfferTimeoutException.class, e.getCause().getClass());
}
- Assert.assertEquals(3, session.getRequestWindow().getSize());
+ Assert.assertEquals(3, session.getSendWindow().getSize());
// now the smsc will send a response back to the second request
simulator0.sendPdu(el1Resp);
@@ -362,7 +362,7 @@ public void windowSizeBlocksAsyncRequest() throws Exception {
future1.await();
// there should be 1 slot free now in the window
- Assert.assertEquals(2, session.getRequestWindow().getSize());
+ Assert.assertEquals(2, session.getSendWindow().getSize());
// this request should now succeed
WindowFuture future3 = session.sendRequestPdu(el3, 3000, true);
@@ -378,7 +378,7 @@ public void windowSizeBlocksAsyncRequest() throws Exception {
future2.await();
future3.await();
- Assert.assertEquals(0, session.getRequestWindow().getSize());
+ Assert.assertEquals(0, session.getSendWindow().getSize());
} finally {
SmppSessionUtil.close(session);
}
@@ -959,5 +959,53 @@ public void receiveUnexpectedPduResponse() throws Exception {
SmppSessionUtil.close(session);
}
}
+
+
+ @Test
+ public void synchronousSendButNeverGetResponse() throws Exception {
+ SmppSessionConfiguration configuration = createDefaultConfiguration();
+ registerServerBindProcessor();
+ clearAllServerSessions();
+
+ // bind and get the simulator session
+ PollableSmppSessionHandler sessionHandler = new PollableSmppSessionHandler();
+ DefaultSmppSession session = (DefaultSmppSession)bootstrap.bind(configuration, sessionHandler);
+
+ SmppSimulatorSessionHandler simulator0 = server.pollNextSession(1000);
+ simulator0.setPduProcessor(null);
+
+ try {
+ try {
+ session.enquireLink(new EnquireLink(), 100);
+ // request should timeout
+ Assert.fail();
+ } catch (SmppTimeoutException e) {
+ // correct behavior
+ }
+
+ // with a "synchronous" type of send, after a timeout, the request
+ // should have been cancelled
+ Assert.assertEquals(0, session.getSendWindow().getSize());
+
+ /**
+ // send a response to a request that was NEVER sent
+ simulator0.sendPdu(el0Resp);
+
+ // we should have received a PDU response
+ PduResponse pdu0 = sessionHandler.getReceivedUnexpectedPduResponses().poll(1000, TimeUnit.MILLISECONDS);
+ Assert.assertNotNull("Unable to receive unexpected PDU response -- perhaps it was routed incorrectly?", pdu0);
+ Assert.assertEquals(SmppConstants.CMD_ID_ENQUIRE_LINK_RESP, pdu0.getCommandId());
+ Assert.assertEquals(0, pdu0.getCommandStatus());
+ Assert.assertEquals(16, pdu0.getCommandLength());
+ Assert.assertEquals(0x1000, pdu0.getSequenceNumber());
+
+ Assert.assertEquals(0, sessionHandler.getReceivedPduRequests().size());
+ Assert.assertEquals(0, sessionHandler.getReceivedExpectedPduResponses().size());
+ Assert.assertEquals(0, sessionHandler.getReceivedUnexpectedPduResponses().size());
+ */
+ } finally {
+ SmppSessionUtil.close(session);
+ }
+ }
}

0 comments on commit 04a9622

Please sign in to comment.