001 /**
002 * Licensed to the Apache Software Foundation (ASF) under one or more
003 * contributor license agreements. See the NOTICE file distributed with
004 * this work for additional information regarding copyright ownership.
005 * The ASF licenses this file to You under the Apache License, Version 2.0
006 * (the "License"); you may not use this file except in compliance with
007 * the License. You may obtain a copy of the License at
008 *
009 * http://www.apache.org/licenses/LICENSE-2.0
010 *
011 * Unless required by applicable law or agreed to in writing, software
012 * distributed under the License is distributed on an "AS IS" BASIS,
013 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
014 * See the License for the specific language governing permissions and
015 * limitations under the License.
016 */
017 package org.apache.camel.component.smpp;
018
019 import java.io.IOException;
020 import java.util.concurrent.locks.ReentrantLock;
021
022 import org.apache.camel.Exchange;
023 import org.apache.camel.Processor;
024 import org.apache.camel.impl.DefaultConsumer;
025 import org.jsmpp.DefaultPDUReader;
026 import org.jsmpp.DefaultPDUSender;
027 import org.jsmpp.SynchronizedPDUSender;
028 import org.jsmpp.bean.AlertNotification;
029 import org.jsmpp.bean.BindType;
030 import org.jsmpp.bean.DataSm;
031 import org.jsmpp.bean.DeliverSm;
032 import org.jsmpp.bean.NumberingPlanIndicator;
033 import org.jsmpp.bean.TypeOfNumber;
034 import org.jsmpp.extra.ProcessRequestException;
035 import org.jsmpp.extra.SessionState;
036 import org.jsmpp.session.BindParameter;
037 import org.jsmpp.session.DataSmResult;
038 import org.jsmpp.session.MessageReceiverListener;
039 import org.jsmpp.session.SMPPSession;
040 import org.jsmpp.session.Session;
041 import org.jsmpp.session.SessionStateListener;
042 import org.jsmpp.util.DefaultComposer;
043 import org.jsmpp.util.MessageIDGenerator;
044 import org.jsmpp.util.MessageId;
045 import org.jsmpp.util.RandomMessageIDGenerator;
046 import org.slf4j.Logger;
047 import org.slf4j.LoggerFactory;
048
049 /**
050 * An implementation of @{link Consumer} which use the SMPP protocol
051 *
052 * @version
053 * @author muellerc
054 */
055 public class SmppConsumer extends DefaultConsumer {
056
057 private static final transient Logger LOG = LoggerFactory.getLogger(SmppConsumer.class);
058
059 private SmppConfiguration configuration;
060 private SMPPSession session;
061 private MessageReceiverListener messageReceiverListener;
062 private SessionStateListener sessionStateListener;
063 private final ReentrantLock reconnectLock = new ReentrantLock();
064
065 /**
066 * The constructor which gets a smpp endpoint, a smpp configuration and a
067 * processor
068 */
069 public SmppConsumer(SmppEndpoint endpoint, SmppConfiguration config, Processor processor) {
070 super(endpoint, processor);
071
072 this.configuration = config;
073 this.sessionStateListener = new SessionStateListener() {
074 public void onStateChange(SessionState newState, SessionState oldState, Object source) {
075 if (newState.equals(SessionState.CLOSED)) {
076 LOG.warn("Loost connection to: " + getEndpoint().getConnectionString()
077 + " - trying to reconnect...");
078 closeSession(session);
079 reconnect(configuration.getInitialReconnectDelay());
080 }
081 }
082 };
083 this.messageReceiverListener = new MessageReceiverListener() {
084 private final MessageIDGenerator messageIDGenerator = new RandomMessageIDGenerator();
085
086 public void onAcceptAlertNotification(AlertNotification alertNotification) {
087 if (LOG.isDebugEnabled()) {
088 LOG.debug("Received an alertNotification " + alertNotification);
089 }
090
091 try {
092 Exchange exchange = getEndpoint().createOnAcceptAlertNotificationExchange(
093 alertNotification);
094
095 LOG.trace("Processing the new smpp exchange...");
096 getProcessor().process(exchange);
097 LOG.trace("Processed the new smpp exchange");
098 } catch (Exception e) {
099 getExceptionHandler().handleException(e);
100 }
101 }
102
103 public void onAcceptDeliverSm(DeliverSm deliverSm) {
104 if (LOG.isDebugEnabled()) {
105 LOG.debug("Received a deliverSm " + deliverSm);
106 }
107
108 try {
109 Exchange exchange = getEndpoint().createOnAcceptDeliverSmExchange(deliverSm);
110
111 LOG.trace("processing the new smpp exchange...");
112 getProcessor().process(exchange);
113 LOG.trace("processed the new smpp exchange");
114 } catch (Exception e) {
115 getExceptionHandler().handleException(e);
116 }
117 }
118
119 public DataSmResult onAcceptDataSm(DataSm dataSm, Session session)
120 throws ProcessRequestException {
121 if (LOG.isDebugEnabled()) {
122 LOG.debug("Received a dataSm " + dataSm);
123 }
124
125 MessageId newMessageId = messageIDGenerator.newMessageId();
126
127 try {
128 Exchange exchange = getEndpoint().createOnAcceptDataSm(dataSm,
129 newMessageId.getValue());
130
131 LOG.trace("processing the new smpp exchange...");
132 getProcessor().process(exchange);
133 LOG.trace("processed the new smpp exchange");
134 } catch (Exception e) {
135 getExceptionHandler().handleException(e);
136 throw new ProcessRequestException(e.getMessage(), 255, e);
137 }
138
139 return new DataSmResult(newMessageId, dataSm.getOptionalParametes());
140 }
141 };
142 }
143
144 @Override
145 protected void doStart() throws Exception {
146 LOG.debug("Connecting to: " + getEndpoint().getConnectionString() + "...");
147
148 super.doStart();
149 session = createSession();
150
151 LOG.info("Connected to: " + getEndpoint().getConnectionString());
152 }
153
154 private SMPPSession createSession() throws IOException {
155 SMPPSession session = createSMPPSession();
156 session.setEnquireLinkTimer(configuration.getEnquireLinkTimer());
157 session.setTransactionTimer(configuration.getTransactionTimer());
158 session.addSessionStateListener(sessionStateListener);
159 session.setMessageReceiverListener(messageReceiverListener);
160 session.connectAndBind(this.configuration.getHost(), this.configuration.getPort(),
161 new BindParameter(BindType.BIND_RX, this.configuration.getSystemId(),
162 this.configuration.getPassword(), this.configuration.getSystemType(),
163 TypeOfNumber.UNKNOWN, NumberingPlanIndicator.UNKNOWN, ""));
164
165 return session;
166 }
167
168 /**
169 * Factory method to easily instantiate a mock SMPPSession
170 *
171 * @return the SMPPSession
172 */
173 SMPPSession createSMPPSession() {
174 if (configuration.getUsingSSL()) {
175 return new SMPPSession(new SynchronizedPDUSender(new DefaultPDUSender(
176 new DefaultComposer())), new DefaultPDUReader(), SmppSSLConnectionFactory
177 .getInstance());
178 } else {
179 return new SMPPSession();
180 }
181 }
182
183 @Override
184 protected void doStop() throws Exception {
185 LOG.debug("Disconnecting from: " + getEndpoint().getConnectionString() + "...");
186
187 super.doStop();
188 closeSession(session);
189
190 LOG.info("Disconnected from: " + getEndpoint().getConnectionString());
191 }
192
193 private void closeSession(SMPPSession session) {
194 if (session != null) {
195 session.removeSessionStateListener(this.sessionStateListener);
196 // remove this hack after http://code.google.com/p/jsmpp/issues/detail?id=93 is fixed
197 try {
198 Thread.sleep(1000);
199 session.unbindAndClose();
200 } catch (Exception e) {
201 LOG.warn("Could not close session " + session);
202 }
203 session = null;
204 }
205 }
206
207 private void reconnect(final long initialReconnectDelay) {
208 if (reconnectLock.tryLock()) {
209 try {
210 Runnable r = new Runnable() {
211 public void run() {
212 boolean reconnected = false;
213
214 LOG.info("Schedule reconnect after " + initialReconnectDelay + " millis");
215 try {
216 Thread.sleep(initialReconnectDelay);
217 } catch (InterruptedException e) {
218 }
219
220 int attempt = 0;
221 while (!(isStopping() || isStopped()) && (session == null || session.getSessionState().equals(SessionState.CLOSED))) {
222 try {
223 LOG.info("Trying to reconnect to " + getEndpoint().getConnectionString() + " - attempt #" + (++attempt) + "...");
224 session = createSession();
225 reconnected = true;
226 } catch (IOException e) {
227 LOG.info("Failed to reconnect to " + getEndpoint().getConnectionString());
228 closeSession(session);
229 try {
230 Thread.sleep(configuration.getReconnectDelay());
231 } catch (InterruptedException ee) {
232 }
233 }
234 }
235
236 if (reconnected) {
237 LOG.info("Reconnected to " + getEndpoint().getConnectionString());
238 }
239 }
240 };
241
242 Thread t = new Thread(r);
243 t.start();
244 t.join();
245 } catch (InterruptedException e) {
246 // noop
247 } finally {
248 reconnectLock.unlock();
249 }
250 }
251 }
252
253 @Override
254 public String toString() {
255 return "SmppConsumer[" + getEndpoint().getConnectionString() + "]";
256 }
257
258 @Override
259 public SmppEndpoint getEndpoint() {
260 return (SmppEndpoint) super.getEndpoint();
261 }
262
263 /**
264 * Returns the smpp configuration
265 *
266 * @return the configuration
267 */
268 public SmppConfiguration getConfiguration() {
269 return configuration;
270 }
271 }