spring-integration-aws/src/main/java/org/springframework/integration/aws/sqs/core/SqsExecutor.java
public Message<?> poll(long timeout) {

		Message<?> message = null;
		String payloadJSON = null;
		com.amazonaws.services.sqs.model.Message qMessage = null;
		int timeoutSeconds = (timeout > 0 ? ((int) (timeout / 1000))
				: receiveMessageWaitTimeout);
		destroyWaitTime = timeoutSeconds;
		try {
			if (queue == null) {
				if (prefetchQueue.isEmpty()) {
					ReceiveMessageRequest request = new ReceiveMessageRequest(
							queueUrl).withWaitTimeSeconds(timeoutSeconds)
							.withMaxNumberOfMessages(prefetchCount)
							.withAttributeNames("All");

					ReceiveMessageResult result = sqsClient
							.receiveMessage(request);
					for (com.amazonaws.services.sqs.model.Message sqsMessage : result
							.getMessages()) {
						prefetchQueue.offer(sqsMessage);
					}
					qMessage = prefetchQueue.poll();
				} else {
					qMessage = prefetchQueue.remove();
				}
				if (qMessage != null) {
					payloadJSON = qMessage.getBody();
					// MD5 verification
					try {
						byte[] computedHash = Md5Utils
								.computeMD5Hash(payloadJSON.getBytes("UTF-8"));
						String hexDigest = new String(
								Hex.encodeHex(computedHash));
						if (!hexDigest.equals(qMessage.getMD5OfBody())) {
							payloadJSON = null; // ignore this message
							log.warn("Dropped message due to MD5 checksum failure");
						}
					} catch (Exception e) {
						log.warn(
								"Failed to verify MD5 checksum: "
										+ e.getMessage(), e);
					}
				}
			} else {
				try {
					payloadJSON = queue.poll(timeoutSeconds, TimeUnit.SECONDS);
				} catch (InterruptedException e) {
					log.warn(e.getMessage(), e);
				}
			}
			if (payloadJSON != null) {
				JSONObject qMessageJSON = new JSONObject(payloadJSON);
				if (qMessageJSON.has(SNS_MESSAGE_KEY)) { // posted from SNS
					payloadJSON = qMessageJSON.getString(SNS_MESSAGE_KEY);
					// XXX: other SNS attributes?
				}
				Message<?> packet = null;
				try {
					packet = messageMarshaller.deserialize(payloadJSON);
				} catch (MessageMarshallerException marshallingException) {
					throw new MessagingException(
							marshallingException.getMessage(),
							marshallingException.getCause());
				}
				MessageBuilder<?> builder = MessageBuilder.fromMessage(packet);
				if (qMessage != null) {
					builder.setHeader(SqsHeaders.MSG_RECEIPT_HANDLE,
							qMessage.getReceiptHandle());
					builder.setHeader(SqsHeaders.AWS_MESSAGE_ID,
							qMessage.getMessageId());
					for (Map.Entry<String, String> e : qMessage.getAttributes()
							.entrySet()) {
						if (e.getKey().equals("ApproximateReceiveCount")) {
							builder.setHeader(SqsHeaders.RECEIVE_COUNT,
									Integer.valueOf(e.getValue()));
						} else if (e.getKey().equals("SentTimestamp")) {
							builder.setHeader(SqsHeaders.SENT_AT,
									new Date(Long.valueOf(e.getValue())));
						} else if (e.getKey().equals(
								"ApproximateFirstReceiveTimestamp")) {
							builder.setHeader(SqsHeaders.FIRST_RECEIVED_AT,
									new Date(Long.valueOf(e.getValue())));
						} else if (e.getKey().equals("SenderId")) {
							builder.setHeader(SqsHeaders.SENDER_AWS_ID,
									e.getValue());
						} else {
							builder.setHeader(e.getKey(), e.getValue());
						}
					}
				} else {
					builder.setHeader(SqsHeaders.MSG_RECEIPT_HANDLE, "");
					// to satisfy test conditions
				}

				message = builder.build();
			}

		} catch (JSONException e) {
			log.warn(e.getMessage(), e);

		} finally {
			destroyWaitTime = 0;
		}

		return message;
	}
public Message<?> poll(long timeout) {

		Message<?> message = null;
		String payloadJSON = null;
		com.amazonaws.services.sqs.model.Message qMessage = null;
		int timeoutSeconds = (timeout > 0 ? ((int) (timeout / 1000))
				: receiveMessageWaitTimeout);
		destroyWaitTime = timeoutSeconds;
		try {
			if (queue == null) {
				if (prefetchQueue.isEmpty()) {
					ReceiveMessageRequest request = new ReceiveMessageRequest(
							queueUrl).withWaitTimeSeconds(timeoutSeconds)
							.withMaxNumberOfMessages(prefetchCount)
							.withAttributeNames("All");

					ReceiveMessageResult result = sqsClient
							.receiveMessage(request);
					for (com.amazonaws.services.sqs.model.Message sqsMessage : result
							.getMessages()) {
						prefetchQueue.offer(sqsMessage);
					}
					qMessage = prefetchQueue.poll();
				} else {
					qMessage = prefetchQueue.remove();
				}
				if (qMessage != null) {
					payloadJSON = qMessage.getBody();
					// MD5 verification
					try {
						byte[] computedHash = Md5Utils
								.computeMD5Hash(payloadJSON.getBytes("UTF-8"));
						String hexDigest = new String(
								Hex.encodeHex(computedHash));
						if (!hexDigest.equals(qMessage.getMD5OfBody())) {
							payloadJSON = null; // ignore this message
							log.warn("Dropped message due to MD5 checksum failure");
						}
					} catch (Exception e) {
						log.warn(
								"Failed to verify MD5 checksum: "
										+ e.getMessage(), e);
					}
				}
			} else {
				try {
					payloadJSON = queue.poll(timeoutSeconds, TimeUnit.SECONDS);
				} catch (InterruptedException e) {
					log.warn(e.getMessage(), e);
				}
			}
			if (payloadJSON != null) {
				JSONObject qMessageJSON = new JSONObject(payloadJSON);
				if (qMessageJSON.has(SNS_MESSAGE_KEY)) { // posted from SNS
					payloadJSON = qMessageJSON.getString(SNS_MESSAGE_KEY);
					// XXX: other SNS attributes?
				}
				Message<?> packet = null;
				try {
					packet = messageMarshaller.deserialize(payloadJSON);
				} catch (MessageMarshallerException marshallingException) {
					throw new MessagingException(
							marshallingException.getMessage(),
							marshallingException.getCause());
				}
				MessageBuilder<?> builder = MessageBuilder.fromMessage(packet);
				if (qMessage != null) {
					builder.setHeader(SqsHeaders.MSG_RECEIPT_HANDLE,
							qMessage.getReceiptHandle());
					builder.setHeader(SqsHeaders.AWS_MESSAGE_ID,
							qMessage.getMessageId());
					for (Map.Entry<String, String> e : qMessage.getAttributes()
							.entrySet()) {
						if (e.getKey().equals("ApproximateReceiveCount")) {
							builder.setHeader(SqsHeaders.RECEIVE_COUNT,
									Integer.valueOf(e.getValue()));
						} else if (e.getKey().equals("SentTimestamp")) {
							builder.setHeader(SqsHeaders.SENT_AT,
									new Date(Long.valueOf(e.getValue())));
						} else if (e.getKey().equals(
								"ApproximateFirstReceiveTimestamp")) {
							builder.setHeader(SqsHeaders.FIRST_RECEIVED_AT,
									new Date(Long.valueOf(e.getValue())));
						} else if (e.getKey().equals("SenderId")) {
							builder.setHeader(SqsHeaders.SENDER_AWS_ID,
									e.getValue());
						} else {
							builder.setHeader(e.getKey(), e.getValue());
						}
					}
				} else {
					builder.setHeader(SqsHeaders.MSG_RECEIPT_HANDLE, "");
					// to satisfy test conditions
				}

				message = builder.build();
			}

		} catch (JSONException e) {
			log.warn(e.getMessage(), e);

		} finally {
			destroyWaitTime = 0;
		}

		return message;
	}
public Message < ? > poll ( long timeout ) { Message < ? > message = null ; String payloadJSON = null ; com . amazonaws . services . sqs . model . Message qMessage = null ; int timeoutSeconds = ( timeout > 0 ? ( ( int ) ( timeout / 1000 ) ) : receiveMessageWaitTimeout ) ; destroyWaitTime = timeoutSeconds ; try { if ( queue == null ) { if ( prefetchQueue . isEmpty ( ) ) { ReceiveMessageRequest request = new ReceiveMessageRequest ( queueUrl ) . withWaitTimeSeconds ( timeoutSeconds ) . withMaxNumberOfMessages ( prefetchCount ) . withAttributeNames ( "All" ) ; ReceiveMessageResult result = sqsClient . receiveMessage ( request ) ; for ( com . amazonaws . services . sqs . model . Message sqsMessage : result . getMessages ( ) ) { prefetchQueue . offer ( sqsMessage ) ; } qMessage = prefetchQueue . poll ( ) ; } else { qMessage = prefetchQueue . remove ( ) ; } if ( qMessage != null ) { payloadJSON = qMessage . getBody ( ) ; // MD5 verification try { byte [ ] computedHash = Md5Utils . computeMD5Hash ( payloadJSON . getBytes ( "UTF-8" ) ) ; String hexDigest = new String ( Hex . encodeHex ( computedHash ) ) ; if ( ! hexDigest . equals ( qMessage . getMD5OfBody ( ) ) ) { payloadJSON = null ; // ignore this message log . warn ( "Dropped message due to MD5 checksum failure" ) ; } } catch ( Exception e ) { log . warn ( "Failed to verify MD5 checksum: " + e . getMessage ( ) , e ) ; } } } else { try { payloadJSON = queue . poll ( timeoutSeconds , TimeUnit . SECONDS ) ; } catch ( InterruptedException e ) { log . warn ( e . getMessage ( ) , e ) ; } } if ( payloadJSON != null ) { JSONObject qMessageJSON = new JSONObject ( payloadJSON ) ; if ( qMessageJSON . has ( SNS_MESSAGE_KEY ) ) { // posted from SNS payloadJSON = qMessageJSON . getString ( SNS_MESSAGE_KEY ) ; // XXX: other SNS attributes? } Message < ? > packet = null ; try { packet = messageMarshaller . deserialize ( payloadJSON ) ; } catch ( MessageMarshallerException marshallingException ) { throw new MessagingException ( marshallingException . getMessage ( ) , marshallingException . getCause ( ) ) ; } MessageBuilder < ? > builder = MessageBuilder . fromMessage ( packet ) ; if ( qMessage != null ) { builder . setHeader ( SqsHeaders . MSG_RECEIPT_HANDLE , qMessage . getReceiptHandle ( ) ) ; builder . setHeader ( SqsHeaders . AWS_MESSAGE_ID , qMessage . getMessageId ( ) ) ; for ( Map . Entry < String , String > e : qMessage . getAttributes ( ) . entrySet ( ) ) { if ( e . getKey ( ) . equals ( "ApproximateReceiveCount" ) ) { builder . setHeader ( SqsHeaders . RECEIVE_COUNT , Integer . valueOf ( e . getValue ( ) ) ) ; } else if ( e . getKey ( ) . equals ( "SentTimestamp" ) ) { builder . setHeader ( SqsHeaders . SENT_AT , new Date ( Long . valueOf ( e . getValue ( ) ) ) ) ; } else if ( e . getKey ( ) . equals ( "ApproximateFirstReceiveTimestamp" ) ) { builder . setHeader ( SqsHeaders . FIRST_RECEIVED_AT , new Date ( Long . valueOf ( e . getValue ( ) ) ) ) ; } else if ( e . getKey ( ) . equals ( "SenderId" ) ) { builder . setHeader ( SqsHeaders . SENDER_AWS_ID , e . getValue ( ) ) ; } else { builder . setHeader ( e . getKey ( ) , e . getValue ( ) ) ; } } } else { builder . setHeader ( SqsHeaders . MSG_RECEIPT_HANDLE , "" ) ; // to satisfy test conditions } message = builder . build ( ) ; } } catch ( JSONException e ) { log . warn ( e . getMessage ( ) , e ) ; } finally { destroyWaitTime = 0 ; } return message ; }
Execute a retrieving (polling) Sqs operation.

@param timeout
time to wait for a message to return.

@return The payload object, which may be null.
Execute a retrieving ( polling ) Sqs operation .