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.impl.DefaultProducer;
024 import org.jsmpp.DefaultPDUReader;
025 import org.jsmpp.DefaultPDUSender;
026 import org.jsmpp.SynchronizedPDUSender;
027 import org.jsmpp.bean.Alphabet;
028 import org.jsmpp.bean.BindType;
029 import org.jsmpp.bean.ESMClass;
030 import org.jsmpp.bean.GeneralDataCoding;
031 import org.jsmpp.bean.MessageClass;
032 import org.jsmpp.bean.NumberingPlanIndicator;
033 import org.jsmpp.bean.RegisteredDelivery;
034 import org.jsmpp.bean.SubmitSm;
035 import org.jsmpp.bean.TypeOfNumber;
036 import org.jsmpp.extra.SessionState;
037 import org.jsmpp.session.BindParameter;
038 import org.jsmpp.session.SMPPSession;
039 import org.jsmpp.session.SessionStateListener;
040 import org.jsmpp.util.DefaultComposer;
041 import org.slf4j.Logger;
042 import org.slf4j.LoggerFactory;
043
044 /**
045 * An implementation of @{link Producer} which use the SMPP protocol
046 *
047 * @version
048 * @author muellerc
049 */
050 public class SmppProducer extends DefaultProducer {
051
052 private static final transient Logger LOG = LoggerFactory.getLogger(SmppProducer.class);
053
054 private SmppConfiguration configuration;
055 private SMPPSession session;
056 private SessionStateListener sessionStateListener;
057 private final ReentrantLock reconnectLock = new ReentrantLock();
058
059 public SmppProducer(SmppEndpoint endpoint, SmppConfiguration config) {
060 super(endpoint);
061 this.configuration = config;
062 this.sessionStateListener = new SessionStateListener() {
063 public void onStateChange(SessionState newState, SessionState oldState, Object source) {
064 if (newState.equals(SessionState.CLOSED)) {
065 LOG.warn("Loosing connection to: " + getEndpoint().getConnectionString() + " - trying to reconnect...");
066 closeSession(session);
067 reconnect(configuration.getInitialReconnectDelay());
068 }
069 }
070 };
071 }
072
073 @Override
074 protected void doStart() throws Exception {
075 LOG.debug("Connecting to: " + getEndpoint().getConnectionString() + "...");
076
077 super.doStart();
078 session = createSession();
079
080 LOG.info("Connected to: " + getEndpoint().getConnectionString());
081 }
082
083 private SMPPSession createSession() throws IOException {
084 SMPPSession session = createSMPPSession();
085 session.setEnquireLinkTimer(this.configuration.getEnquireLinkTimer());
086 session.setTransactionTimer(this.configuration.getTransactionTimer());
087 session.addSessionStateListener(sessionStateListener);
088 session.connectAndBind(
089 this.configuration.getHost(),
090 this.configuration.getPort(),
091 new BindParameter(
092 BindType.BIND_TX,
093 this.configuration.getSystemId(),
094 this.configuration.getPassword(),
095 this.configuration.getSystemType(),
096 TypeOfNumber.valueOf(configuration.getTypeOfNumber()),
097 NumberingPlanIndicator.valueOf(configuration.getNumberingPlanIndicator()),
098 ""));
099
100 return session;
101 }
102
103 /**
104 * Factory method to easily instantiate a mock SMPPSession
105 *
106 * @return the SMPPSession
107 */
108 SMPPSession createSMPPSession() {
109 if (configuration.getUsingSSL()) {
110 return new SMPPSession(new SynchronizedPDUSender(new DefaultPDUSender(new DefaultComposer())),
111 new DefaultPDUReader(), SmppSSLConnectionFactory.getInstance());
112 } else {
113 return new SMPPSession();
114 }
115 }
116
117 public void process(Exchange exchange) throws Exception {
118 if (LOG.isDebugEnabled()) {
119 LOG.debug("Sending a short message for exchange id '"
120 + exchange.getExchangeId() + "'...");
121 }
122
123 // only possible by trying to reconnect
124 if (this.session == null) {
125 throw new IOException("Lost connection to " + getEndpoint().getConnectionString() + " and yet not reconnected");
126 }
127
128 SubmitSm submitSm = getEndpoint().getBinding().createSubmitSm(exchange);
129 String messageId = session.submitShortMessage(
130 submitSm.getServiceType(),
131 TypeOfNumber.valueOf(submitSm.getSourceAddrTon()),
132 NumberingPlanIndicator.valueOf(submitSm.getSourceAddrNpi()),
133 submitSm.getSourceAddr(),
134 TypeOfNumber.valueOf(submitSm.getDestAddrTon()),
135 NumberingPlanIndicator.valueOf(submitSm.getDestAddrNpi()),
136 submitSm.getDestAddress(),
137 new ESMClass(),
138 submitSm.getProtocolId(),
139 submitSm.getPriorityFlag(),
140 submitSm.getScheduleDeliveryTime(),
141 submitSm.getValidityPeriod(),
142 new RegisteredDelivery(submitSm.getRegisteredDelivery()),
143 submitSm.getReplaceIfPresent(),
144 new GeneralDataCoding(
145 false,
146 false,
147 MessageClass.CLASS1,
148 Alphabet.valueOf(submitSm.getDataCoding())),
149 (byte) 0,
150 submitSm.getShortMessage());
151
152 if (LOG.isDebugEnabled()) {
153 LOG.debug("Sent a short message for exchange id '"
154 + exchange.getExchangeId() + "' and received message id '"
155 + messageId + "'");
156 }
157
158 if (exchange.getPattern().isOutCapable()) {
159 if (LOG.isDebugEnabled()) {
160 LOG.debug("Exchange is out capable, setting headers on out exchange...");
161 }
162 exchange.getOut().setHeader(SmppBinding.ID, messageId);
163 } else {
164 if (LOG.isDebugEnabled()) {
165 LOG.debug("Exchange is not out capable, setting headers on in exchange...");
166 }
167 exchange.getIn().setHeader(SmppBinding.ID, messageId);
168 }
169 }
170
171 @Override
172 protected void doStop() throws Exception {
173 LOG.debug("Disconnecting from: " + getEndpoint().getConnectionString() + "...");
174
175 super.doStop();
176 closeSession(session);
177
178 LOG.info("Disconnected from: " + getEndpoint().getConnectionString());
179 }
180
181 private void closeSession(SMPPSession session) {
182 if (session != null) {
183 session.removeSessionStateListener(this.sessionStateListener);
184 // remove this hack after http://code.google.com/p/jsmpp/issues/detail?id=93 is fixed
185 try {
186 Thread.sleep(1000);
187 session.unbindAndClose();
188 } catch (Exception e) {
189 LOG.warn("Could not close session " + session);
190 }
191 session = null;
192 }
193 }
194
195 private void reconnect(final long initialReconnectDelay) {
196 if (reconnectLock.tryLock()) {
197 try {
198 Runnable r = new Runnable() {
199 public void run() {
200 boolean reconnected = false;
201
202 LOG.info("Schedule reconnect after " + initialReconnectDelay + " millis");
203 try {
204 Thread.sleep(initialReconnectDelay);
205 } catch (InterruptedException e) {
206 }
207
208 int attempt = 0;
209 while (!(isStopping() || isStopped()) && (session == null || session.getSessionState().equals(SessionState.CLOSED))) {
210 try {
211 LOG.info("Trying to reconnect to " + getEndpoint().getConnectionString() + " - attempt #" + (++attempt) + "...");
212 session = createSession();
213 reconnected = true;
214 } catch (IOException e) {
215 LOG.info("Failed to reconnect to " + getEndpoint().getConnectionString());
216 closeSession(session);
217 try {
218 Thread.sleep(configuration.getReconnectDelay());
219 } catch (InterruptedException ee) {
220 }
221 }
222 }
223
224 if (reconnected) {
225 LOG.info("Reconnected to " + getEndpoint().getConnectionString());
226 }
227 }
228 };
229
230 Thread t = new Thread(r);
231 t.start();
232 t.join();
233 } catch (InterruptedException e) {
234 // noop
235 } finally {
236 reconnectLock.unlock();
237 }
238 }
239 }
240
241 @Override
242 public SmppEndpoint getEndpoint() {
243 return (SmppEndpoint) super.getEndpoint();
244 }
245
246 /**
247 * Returns the smppConfiguration for this producer
248 *
249 * @return the configuration
250 */
251 public SmppConfiguration getConfiguration() {
252 return configuration;
253 }
254
255 @Override
256 public String toString() {
257 return "SmppProducer[" + getEndpoint().getConnectionString() + "]";
258 }
259 }