1 package com.eviware.soapui.impl.wsdl.submit.transports.jms;
2
3 import hermes.Hermes;
4
5 import java.io.BufferedReader;
6 import java.io.File;
7 import java.io.FileOutputStream;
8 import java.io.IOException;
9 import java.io.InputStream;
10 import java.io.InputStreamReader;
11 import java.io.OutputStream;
12 import java.util.ArrayList;
13 import java.util.Calendar;
14 import java.util.List;
15
16 import javax.jms.BytesMessage;
17 import javax.jms.Connection;
18 import javax.jms.Destination;
19 import javax.jms.JMSException;
20 import javax.jms.MapMessage;
21 import javax.jms.Message;
22 import javax.jms.MessageConsumer;
23 import javax.jms.MessageProducer;
24 import javax.jms.Queue;
25 import javax.jms.Session;
26 import javax.jms.TextMessage;
27 import javax.jms.Topic;
28 import javax.jms.TopicSubscriber;
29 import javax.naming.NamingException;
30
31 import org.apache.commons.io.output.ByteArrayOutputStream;
32 import org.apache.commons.lang.NotImplementedException;
33
34 import com.eviware.soapui.SoapUI;
35 import com.eviware.soapui.impl.rest.RestRequest;
36 import com.eviware.soapui.impl.support.AbstractHttpRequest;
37 import com.eviware.soapui.impl.wsdl.WsdlProject;
38 import com.eviware.soapui.impl.wsdl.WsdlRequest;
39 import com.eviware.soapui.impl.wsdl.submit.RequestFilter;
40 import com.eviware.soapui.impl.wsdl.submit.RequestTransport;
41 import com.eviware.soapui.impl.wsdl.submit.RequestTransportRegistry.CannotResolveJmsTypeException;
42 import com.eviware.soapui.impl.wsdl.submit.RequestTransportRegistry.MissingTransportException;
43 import com.eviware.soapui.impl.wsdl.submit.transports.http.BaseHttpRequestTransport;
44 import com.eviware.soapui.impl.wsdl.submit.transports.jms.util.HermesUtils;
45 import com.eviware.soapui.impl.wsdl.submit.transports.jms.util.JMSUtils;
46 import com.eviware.soapui.impl.wsdl.support.RequestFileAttachment;
47 import com.eviware.soapui.impl.wsdl.support.jms.header.JMSHeaderConfig;
48 import com.eviware.soapui.impl.wsdl.teststeps.HttpTestRequest;
49 import com.eviware.soapui.model.iface.Attachment;
50 import com.eviware.soapui.model.iface.Request;
51 import com.eviware.soapui.model.iface.Response;
52 import com.eviware.soapui.model.iface.SubmitContext;
53 import com.eviware.soapui.model.propertyexpansion.PropertyExpander;
54 import com.eviware.soapui.model.support.ModelSupport;
55 import com.eviware.soapui.support.StringUtils;
56 import com.eviware.soapui.support.xml.XmlUtils;
57
58 public class HermesJmsRequestTransport implements RequestTransport
59 {
60
61 public static final String IS_JMS_MESSAGE_RECEIVED = "JMS_MESSAGE_RECEIVE";
62 public static final String JMS_MESSAGE_SEND = "JMS_MESSAGE_SEND";
63 public static final String JMS_RESPONSE = "JMS_RESPONSE";
64 public static final String JMS_ERROR = "JMS_ERROR";
65 public static final String JMS_RECEIVE_TIMEOUT = "JMS_RECEIVE_TIMEOUT";
66
67 protected String username;
68 protected String password;
69 protected JMSEndpoint jmsEndpoint;
70 protected String durableSubscriptionName;
71 protected String clientID;
72 protected String messageSelector;
73 protected boolean sendAsBytesMessage;
74 protected boolean addSoapAction;
75 protected Hermes hermes;
76 protected static List<RequestFilter> filters = new ArrayList<RequestFilter>();
77
78 public Response sendRequest( SubmitContext submitContext, Request request ) throws Exception
79 {
80 long timeStarted = Calendar.getInstance().getTimeInMillis();
81 submitContext.setProperty( JMS_RECEIVE_TIMEOUT, getTimeout( submitContext, request ) );
82
83 return resolveType( submitContext, request ).execute( submitContext, request, timeStarted );
84 }
85
86 protected void init( SubmitContext submitContext, Request request ) throws NamingException
87 {
88 this.jmsEndpoint = new JMSEndpoint( request, submitContext );
89 this.hermes = getHermes( jmsEndpoint.getSessionName(), request );
90 this.username = submitContext.expand( request.getUsername() );
91 this.password = submitContext.expand( request.getPassword() );
92 JMSHeaderConfig jmsConfig = ( ( AbstractHttpRequest<?> )request ).getJMSHeaderConfig();
93 this.durableSubscriptionName = submitContext.expand( jmsConfig.getDurableSubscriptionName() );
94 this.clientID = submitContext.expand( jmsConfig.getClientID() );
95 this.messageSelector = jmsConfig.getMessageSelector();
96
97 this.sendAsBytesMessage = jmsConfig.getSendAsBytesMessage();
98 this.addSoapAction = jmsConfig.getSoapActionAdd();
99 submitContext.setProperty( HermesJmsRequestTransport.IS_JMS_MESSAGE_RECEIVED, false );
100 }
101
102 protected Response execute( SubmitContext submitContext, Request request, long timeStarted ) throws Exception
103 {
104 throw new NotImplementedException();
105 }
106
107 private HermesJmsRequestTransport resolveType( SubmitContext submitContext, Request request )
108 throws CannotResolveJmsTypeException, MissingTransportException
109 {
110 String endpoint = PropertyExpander.expandProperties( submitContext, request.getEndpoint() );
111 int ix = endpoint.indexOf( "://" );
112 if( ix == -1 )
113 throw new MissingTransportException( "Missing protocol in endpoint [" + endpoint + "]" );
114
115 String[] params = JMSEndpoint.extractEndpointParameters( request, submitContext );
116
117
118 if( params.length == 2 )
119 {
120 String destinationName = PropertyExpander.expandProperties( submitContext, params[1] );
121 if( destinationName.startsWith( JMSEndpoint.QUEUE_ENDPOINT_PREFIX ) )
122 {
123 return new HermesJmsRequestSendTransport();
124 }
125 else if( destinationName.startsWith( JMSEndpoint.TOPIC_ENDPOINT_PREFIX ) )
126 {
127 return new HermesJmsRequestPublishTransport();
128 }
129 else
130 {
131 cannotResolve();
132 }
133
134 }
135
136 else if( params.length == 3 && PropertyExpander.expandProperties( submitContext, params[1] ).equals( "-" ) )
137 {
138 String destinationName = PropertyExpander.expandProperties( submitContext, params[2] );
139 if( destinationName.startsWith( JMSEndpoint.QUEUE_ENDPOINT_PREFIX ) )
140 {
141 return new HermesJmsRequestReceiveTransport();
142 }
143 else if( destinationName.startsWith( JMSEndpoint.TOPIC_ENDPOINT_PREFIX ) )
144 {
145 return new HermesJmsRequestSubscribeTransport();
146 }
147 else
148 {
149 cannotResolve();
150 }
151 }
152
153 else if( params.length == 3 )
154 {
155 String destinationSendName = PropertyExpander.expandProperties( submitContext, params[1] );
156 String destinationReceiveName = PropertyExpander.expandProperties( submitContext, params[2] );
157 if( destinationSendName.startsWith( JMSEndpoint.QUEUE_ENDPOINT_PREFIX )
158 && destinationReceiveName.startsWith( JMSEndpoint.QUEUE_ENDPOINT_PREFIX ) )
159 {
160 return new HermesJmsRequestSendReceiveTransport();
161 }
162 else if( destinationSendName.startsWith( JMSEndpoint.QUEUE_ENDPOINT_PREFIX )
163 && destinationReceiveName.startsWith( JMSEndpoint.TOPIC_ENDPOINT_PREFIX ) )
164 {
165 return new HermesJmsRequestSendSubscribeTransport();
166 }
167 else if( destinationSendName.startsWith( JMSEndpoint.TOPIC_ENDPOINT_PREFIX )
168 && destinationReceiveName.startsWith( JMSEndpoint.TOPIC_ENDPOINT_PREFIX ) )
169 {
170 return new HermesJmsRequestPublishSubscribeTransport();
171 }
172 else if( destinationSendName.startsWith( JMSEndpoint.TOPIC_ENDPOINT_PREFIX )
173 && destinationReceiveName.startsWith( JMSEndpoint.QUEUE_ENDPOINT_PREFIX ) )
174 {
175 return new HermesJmsRequestPublishReceiveTransport();
176 }
177 else
178 {
179 cannotResolve();
180 }
181 }
182 else
183 {
184 cannotResolve();
185 }
186 return null;
187 }
188
189 private static void cannotResolve() throws CannotResolveJmsTypeException
190 {
191 throw new CannotResolveJmsTypeException(
192 "\nBad jms alias! \nFor JMS please use this endpont pattern:\nfor sending 'jms://sessionName::queue_myqueuename' \nfor receive 'jms://sessionName::-::queue_myqueuename'\nfor send-receive 'jms://sessionName::queue_myqueuename1::queue_myqueuename2'" );
193 }
194
195 protected Hermes getHermes( String sessionName, Request request ) throws NamingException
196 {
197 WsdlProject project = ( WsdlProject )ModelSupport.getModelItemProject( request );
198 return HermesUtils.getHermes( project, sessionName );
199 }
200
201 protected long getTimeout( SubmitContext submitContext, Request request )
202 {
203 String timeout = PropertyExpander.expandProperties( submitContext, request.getTimeout() );
204 long to = 0;
205 try
206 {
207 to = Long.parseLong( timeout );
208 }
209 catch( Exception e )
210 {
211 }
212
213 return to;
214 }
215
216 protected JMSHeader createJMSHeader( SubmitContext submitContext, Request request, Hermes hermes, Message message, Destination replyToDestination )
217 {
218 JMSHeader jmsHeader = new JMSHeader();
219 jmsHeader.setMessageHeaders( message, request, hermes, submitContext );
220 JMSHeader.setMessageProperties( message, request, hermes, submitContext );
221 try
222 {
223 if (message.getJMSReplyTo() == null)
224 {
225 message.setJMSReplyTo(replyToDestination);
226 }
227
228 if( addSoapAction )
229 {
230 message.setStringProperty( JMSHeader.SOAPJMS_SOAP_ACTION, request.getOperation().getName() );
231 message.setStringProperty( JMSHeader.SOAP_ACTION, request.getOperation().getName() );
232 }
233 }
234 catch( JMSException e )
235 {
236 SoapUI.logError( e );
237 }
238 return jmsHeader;
239 }
240
241 protected void closeSessionAndConnection( Connection connection, Session session ) throws JMSException
242 {
243 if( session != null )
244 session.close();
245 if( connection != null )
246 connection.close();
247 }
248
249 protected Response errorResponse( SubmitContext submitContext, Request request, long timeStarted, JMSException jmse )
250 {
251 JMSResponse response;
252 SoapUI.logError( jmse );
253 submitContext.setProperty( JMS_ERROR, jmse );
254 response = new JMSResponse( "", null, null, request, timeStarted );
255 submitContext.setProperty( JMS_RESPONSE, response );
256 return response;
257 }
258
259 protected Message messageSend( SubmitContext submitContext, Request request, Session session, Hermes hermes,
260 Queue queueSend, Destination replyToDestination ) throws JMSException
261 {
262 MessageProducer messageProducer = session.createProducer( queueSend );
263 Message messageSend = createMessage( submitContext, request, session );
264 return send( submitContext, request, hermes, messageProducer, messageSend, replyToDestination );
265 }
266
267 protected Message messagePublish( SubmitContext submitContext, Request request, Session topicSession, Hermes hermes,
268 Topic topicPublish, Destination replyToDestination ) throws JMSException
269 {
270 MessageProducer topicPublisher = topicSession.createProducer( topicPublish );
271 Message messagePublish = createMessage( submitContext, request, topicSession );
272 return send( submitContext, request, hermes, topicPublisher, messagePublish, replyToDestination );
273 }
274
275 private Message send( SubmitContext submitContext, Request request, Hermes hermes, MessageProducer messageProducer,
276 Message message, Destination replyToDestination ) throws JMSException
277 {
278 JMSHeader jmsHeader = createJMSHeader( submitContext, request, hermes, message, replyToDestination );
279 messageProducer.send( message, message.getJMSDeliveryMode(), message.getJMSPriority(), jmsHeader.getTimeTolive() );
280 submitContext.setProperty( JMS_MESSAGE_SEND, message );
281 return message;
282 }
283
284 protected Response makeResponse( SubmitContext submitContext, Request request, long timeStarted,
285 Message messageSend, MessageConsumer messageConsumer ) throws JMSException
286 {
287 long timeout = getTimeout( submitContext, request );
288 Message messageReceive = messageConsumer.receive( timeout );
289 if( messageReceive != null )
290 {
291 JMSResponse response = resolveMessage( request, timeStarted, messageSend, messageReceive );
292 submitContext.setProperty( IS_JMS_MESSAGE_RECEIVED, true );
293 submitContext.setProperty( JMS_RESPONSE, response );
294 return response;
295 }
296 else
297 {
298 return new JMSResponse( "", null, null, request, timeStarted );
299 }
300 }
301
302 private JMSResponse resolveMessage( Request request, long timeStarted, Message messageSend, Message messageReceive )
303 throws JMSException
304 {
305 if( messageReceive instanceof TextMessage )
306 {
307 TextMessage textMessageReceive = ( TextMessage )messageReceive;
308 return new JMSResponse( textMessageReceive.getText(), messageSend, textMessageReceive, request, timeStarted );
309 }
310 else if( messageReceive instanceof MapMessage )
311 {
312 MapMessage mapMessageReceive = ( MapMessage )messageReceive;
313 return new JMSResponse( JMSUtils.extractMapMessagePayloadToXML( mapMessageReceive ), messageSend,
314 mapMessageReceive, request, timeStarted );
315 }
316 else if( messageReceive instanceof BytesMessage )
317 {
318
319 BytesMessage bytesMessageReceive = ( BytesMessage )messageReceive;
320
321 String bytesMessageAsString = new String( JMSUtils.extractByteArrayFromMessage( bytesMessageReceive ) );
322
323 if( XmlUtils.seemsToBeXml( bytesMessageAsString ) )
324 {
325 return new JMSResponse( bytesMessageAsString, messageSend, bytesMessageReceive, request, timeStarted );
326 }
327 else
328 {
329 JMSResponse jmsResponse = new JMSResponse( "", messageSend, bytesMessageReceive, request, timeStarted );
330 addAttachment( request, bytesMessageReceive, jmsResponse );
331 return jmsResponse;
332 }
333 }
334 return null;
335 }
336
337 protected Response makeEmptyResponse( SubmitContext submitContext, Request request, long timeStarted,
338 Message messageSend )
339 {
340 JMSResponse response = new JMSResponse( "", messageSend, null, request, timeStarted );
341 submitContext.setProperty( JMS_RESPONSE, response );
342 return response;
343 }
344
345 private Message createMessage( SubmitContext submitContext, Request request, Session session ) throws JMSException
346 {
347 if( request instanceof WsdlRequest || request instanceof HttpTestRequest || request instanceof RestRequest )
348 {
349 if( hasAttachment( request ) )
350 {
351 if( isTextAttachment( request ) && !sendAsBytesMessage )
352 {
353 return createTextMessageFromAttachment( submitContext, request, session );
354 }
355 else
356 {
357 return createBytesMessage( request, session );
358 }
359 }
360 else
361 {
362 String requestContent = applyFilters( submitContext, request );
363 if( sendAsBytesMessage )
364 {
365 return createBytesMessageFromText( submitContext, requestContent, session );
366 }
367 else
368 {
369 return createTextMessage( submitContext, requestContent, session );
370 }
371 }
372 }
373
374 return null;
375 }
376
377 private String applyFilters( SubmitContext submitContext, Request request )
378 {
379 submitContext.setProperty( BaseHttpRequestTransport.REQUEST_CONTENT, request.getRequestContent() );
380 submitContext.setProperty( WSDL_REQUEST, request );
381
382 for( RequestFilter filter : filters )
383 {
384 filter.filterRequest( submitContext, request );
385 }
386
387 String requestContent = ( String )submitContext.getProperty( BaseHttpRequestTransport.REQUEST_CONTENT );
388 return requestContent;
389 }
390
391 private Message createBytesMessageFromText( SubmitContext submitContext, String requestContent, Session session )
392 throws JMSException
393 {
394 BytesMessage bytesMessage = session.createBytesMessage();
395 bytesMessage.writeBytes( requestContent.getBytes() );
396 return bytesMessage;
397 }
398
399 private Message createTextMessageFromAttachment( SubmitContext submitContext, Request request, Session session )
400 {
401 try
402 {
403 String content = convertStreamToString( request.getAttachments()[0].getInputStream() );
404 TextMessage textMessageSend = session.createTextMessage();
405 String messageBody = PropertyExpander.expandProperties( submitContext, content );
406 textMessageSend.setText( messageBody );
407 return textMessageSend;
408 }
409 catch( Exception e )
410 {
411 SoapUI.logError( e );
412 }
413 return null;
414 }
415
416 private String convertStreamToString( InputStream is )
417 {
418 BufferedReader reader = new BufferedReader( new InputStreamReader( is ) );
419 StringBuilder sb = new StringBuilder();
420
421 String line = null;
422 try
423 {
424 while( ( line = reader.readLine() ) != null )
425 {
426 sb.append( line + "\n" );
427 }
428 }
429 catch( IOException e )
430 {
431 e.printStackTrace();
432 }
433 finally
434 {
435 try
436 {
437 is.close();
438 }
439 catch( IOException e )
440 {
441 e.printStackTrace();
442 }
443 }
444 return sb.toString();
445 }
446
447 private boolean hasAttachment( Request request )
448 {
449 if( request.getAttachments().length > 0 )
450 return true;
451 return false;
452 }
453
454 private Message createTextMessage( SubmitContext submitContext, String requestContent, Session session )
455 throws JMSException
456 {
457 TextMessage textMessageSend = session.createTextMessage();
458 textMessageSend.setText( requestContent );
459 return textMessageSend;
460 }
461
462 private boolean isTextAttachment( Request request )
463 {
464 if( request.getAttachments().length > 0
465 && ( request.getAttachments()[0].getContentType().contains( "/text" )
466 || request.getAttachments()[0].getContentType().contains( "/xml" ) || request.getAttachments()[0]
467 .getContentType().contains( "text/plain" ) ) )
468 {
469 return true;
470 }
471 return false;
472 }
473
474 private Message createBytesMessage( Request request, Session session )
475 {
476 try
477 {
478 InputStream in = request.getAttachments()[0].getInputStream();
479 int buff = -1;
480 ByteArrayOutputStream baos = new ByteArrayOutputStream();
481 while( ( buff = in.read() ) != -1 )
482 {
483 baos.write( buff );
484 }
485 BytesMessage bytesMessage = session.createBytesMessage();
486 bytesMessage.writeBytes( baos.toByteArray() );
487 return bytesMessage;
488 }
489 catch( Exception e )
490 {
491 SoapUI.logError( e );
492 }
493 return null;
494 }
495
496 private void addAttachment( Request request, BytesMessage bytesMessageReceive, JMSResponse jmsResponse )
497 throws JMSException
498 {
499 try
500 {
501 byte[] buff = new byte[1];
502 File temp = File.createTempFile( "bytesmessage", ".tmp" );
503 OutputStream out = new FileOutputStream( temp );
504 bytesMessageReceive.reset();
505 while( bytesMessageReceive.readBytes( buff ) != -1 )
506 {
507 out.write( buff );
508 }
509 out.close();
510 Attachment[] attachments = new Attachment[] { new RequestFileAttachment( temp, false,
511 ( AbstractHttpRequest<?> )request ) };
512 jmsResponse.setAttachments( attachments );
513 }
514 catch( IOException e )
515 {
516 SoapUI.logError( e );
517 }
518 }
519
520 protected TopicSubscriber createDurableSubscription( SubmitContext submitContext, Session topicSession,
521 JMSConnectionHolder jmsConnectionHolder ) throws JMSException, NamingException
522 {
523
524 Topic topicSubscribe = jmsConnectionHolder.getTopic( jmsConnectionHolder.getJmsEndpoint().getReceive() );
525
526
527 TopicSubscriber topicDurableSubsriber = topicSession.createDurableSubscriber( topicSubscribe, StringUtils
528 .hasContent( durableSubscriptionName ) ? durableSubscriptionName : "durableSubscription"
529 + jmsConnectionHolder.getJmsEndpoint().getReceive(), submitContext.expand( messageSelector ), false );
530 return topicDurableSubsriber;
531 }
532
533 @SuppressWarnings( "serial" )
534 public static class UnresolvedJMSEndpointException extends Exception
535 {
536 public UnresolvedJMSEndpointException( String msg )
537 {
538 super( msg );
539 }
540 }
541
542 public void abortRequest( SubmitContext submitContext )
543 {
544 }
545
546 public void addRequestFilter( RequestFilter filter )
547 {
548 filters.add( filter );
549 }
550
551 public void removeRequestFilter( RequestFilter filter )
552 {
553 filters.remove( filter );
554 }
555 }