| /* |
| * |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you 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 org.apache.qpid.client; |
| |
| import java.io.IOException; |
| import java.nio.ByteBuffer; |
| import java.security.GeneralSecurityException; |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.List; |
| import java.util.UUID; |
| |
| import javax.crypto.spec.SecretKeySpec; |
| import javax.jms.JMSException; |
| import javax.jms.Message; |
| import javax.jms.Queue; |
| import javax.jms.Topic; |
| |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import org.apache.qpid.QpidException; |
| import org.apache.qpid.client.failover.FailoverException; |
| import org.apache.qpid.client.message.AMQMessageDelegate_0_8; |
| import org.apache.qpid.client.message.AbstractJMSMessage; |
| import org.apache.qpid.client.message.Encrypted091MessageFactory; |
| import org.apache.qpid.client.message.MessageEncryptionHelper; |
| import org.apache.qpid.client.message.QpidMessageProperties; |
| import org.apache.qpid.client.protocol.BlockingMethodFrameListener; |
| import org.apache.qpid.client.util.JMSExceptionHelper; |
| import org.apache.qpid.configuration.ClientProperties; |
| import org.apache.qpid.framing.AMQFrame; |
| import org.apache.qpid.framing.AMQMethodBody; |
| import org.apache.qpid.framing.BasicAckBody; |
| import org.apache.qpid.framing.BasicContentHeaderProperties; |
| import org.apache.qpid.framing.BasicNackBody; |
| import org.apache.qpid.framing.BasicPublishBody; |
| import org.apache.qpid.framing.CompositeAMQDataBlock; |
| import org.apache.qpid.framing.ContentBody; |
| import org.apache.qpid.framing.ContentHeaderBody; |
| import org.apache.qpid.framing.ExchangeDeclareBody; |
| import org.apache.qpid.framing.FieldTable; |
| import org.apache.qpid.framing.MethodRegistry; |
| import org.apache.qpid.util.GZIPUtils; |
| |
| public class BasicMessageProducer_0_8 extends BasicMessageProducer |
| { |
| private static final Logger _logger = LoggerFactory.getLogger(BasicMessageProducer_0_8.class); |
| private static final boolean SET_EXPIRATION_AS_TTL = Boolean.getBoolean(ClientProperties.SET_EXPIRATION_AS_TTL); |
| |
| BasicMessageProducer_0_8(AMQConnection connection, AMQDestination destination, boolean transacted, int channelId, |
| AMQSession session, AMQProtocolHandler protocolHandler, long producerId, Boolean immediate, Boolean mandatory) throws |
| QpidException |
| { |
| super(_logger,connection, destination,transacted,channelId,session, producerId, immediate, mandatory); |
| } |
| |
| void declareDestination(AMQDestination destination) throws QpidException |
| { |
| |
| if (destination.getDestSyntax() == AMQDestination.DestSyntax.ADDR) |
| { |
| getSession().resolveAddress(destination, false, false); |
| |
| getSession().handleLinkCreation(destination); |
| getSession().sync(); |
| } |
| else |
| { |
| if (getSession().isDeclareExchanges() && !getSession().isResolved(destination)) |
| { |
| final MethodRegistry methodRegistry = getSession().getMethodRegistry(); |
| ExchangeDeclareBody body = |
| methodRegistry.createExchangeDeclareBody(getSession().getTicket(), |
| destination.getExchangeName(), |
| destination.getExchangeClass(), |
| destination.getExchangeName() |
| .startsWith("amq."), |
| destination.isExchangeDurable(), |
| destination.isExchangeAutoDelete(), |
| destination.isExchangeInternal(), |
| true, |
| null); |
| AMQFrame declare = body.generateFrame(getChannelId()); |
| |
| getConnection().getProtocolHandler().writeFrame(declare); |
| getSession().setResolved(destination); |
| } |
| } |
| } |
| |
| void sendMessage(AMQDestination destination, Message origMessage, AbstractJMSMessage message, |
| UUID messageId, int deliveryMode, int priority, long timeToLive, boolean mandatory, |
| boolean immediate, final long deliveryDelay) throws JMSException |
| { |
| |
| |
| AMQMessageDelegate_0_8 delegate = (AMQMessageDelegate_0_8) message.getDelegate(); |
| BasicContentHeaderProperties contentHeaderProperties = delegate.getContentHeaderProperties(); |
| |
| String routingKey = destination.getRoutingKey(); |
| |
| FieldTable headers = delegate.getContentHeaderProperties().getHeaders(); |
| |
| if (destination.getDestSyntax() == AMQDestination.DestSyntax.ADDR && |
| (destination.getSubject() != null |
| || (headers != null && headers.get(QpidMessageProperties.QPID_SUBJECT) != null))) |
| { |
| |
| if (headers.get(QpidMessageProperties.QPID_SUBJECT) == null) |
| { |
| // use default subject in address string |
| headers.setString(QpidMessageProperties.QPID_SUBJECT, destination.getSubject()); |
| } |
| |
| if (destination.getAddressType() == AMQDestination.TOPIC_TYPE) |
| { |
| routingKey = headers.getString(QpidMessageProperties.QPID_SUBJECT); |
| } |
| } |
| |
| BasicPublishBody body = getSession().getMethodRegistry().createBasicPublishBody(getSession().getTicket(), |
| destination.getExchangeName(), |
| routingKey, |
| mandatory, |
| immediate); |
| |
| AMQFrame publishFrame = body.generateFrame(getChannelId()); |
| |
| message.prepareForSending(); |
| ByteBuffer payload = message.getData(); |
| |
| contentHeaderProperties.setUserId(getUserID()); |
| |
| //Set the JMS_QPID_DESTTYPE for 0-8/9 messages |
| int type; |
| if (destination instanceof Topic) |
| { |
| type = AMQDestination.TOPIC_TYPE; |
| } |
| else if (destination instanceof Queue) |
| { |
| type = AMQDestination.QUEUE_TYPE; |
| } |
| else |
| { |
| type = AMQDestination.UNKNOWN_TYPE; |
| } |
| |
| //Set JMS_QPID_DESTTYPE |
| delegate.getContentHeaderProperties() |
| .getHeaders() |
| .setInteger(CustomJMSXProperty.JMS_QPID_DESTTYPE.getShortStringName(), type); |
| |
| long currentTime; |
| if (!isDisableTimestamps()) |
| { |
| |
| currentTime = System.currentTimeMillis(); |
| contentHeaderProperties.setTimestamp(currentTime); |
| |
| if (timeToLive > 0) |
| { |
| if (!SET_EXPIRATION_AS_TTL) |
| { |
| //default behaviour used by Qpid |
| contentHeaderProperties.setExpiration(currentTime + timeToLive); |
| } |
| else |
| { |
| //alternative behaviour for brokers interpreting the expiration header directly as a TTL. |
| contentHeaderProperties.setExpiration(timeToLive); |
| } |
| } |
| else |
| { |
| contentHeaderProperties.setExpiration(0); |
| } |
| } |
| else |
| { |
| currentTime = 0L; |
| } |
| |
| if(deliveryDelay != 0L && headers.get(QpidMessageProperties.QPID_NOT_VALID_BEFORE) == null) |
| { |
| if(currentTime == 0L) |
| { |
| currentTime = System.currentTimeMillis(); |
| } |
| headers.setLong(QpidMessageProperties.QPID_NOT_VALID_BEFORE, deliveryDelay+currentTime); |
| } |
| |
| |
| contentHeaderProperties.setDeliveryMode((byte) deliveryMode); |
| contentHeaderProperties.setPriority((byte) priority); |
| |
| int size = (payload != null) ? payload.remaining() : 0; |
| AMQFrame contentHeaderFrame; |
| final AMQFrame[] frames; |
| boolean encrypt = message.getBooleanProperty(MessageEncryptionHelper.ENCRYPT_HEADER) || destination.sendEncrypted(); |
| if(encrypt) |
| { |
| MessageEncryptionHelper encryptionHelper = getSession().getMessageEncryptionHelper(); |
| try |
| { |
| SecretKeySpec secretKey = encryptionHelper.createSecretKey(); |
| |
| contentHeaderProperties.getHeaders().remove(MessageEncryptionHelper.ENCRYPT_HEADER); |
| |
| String recipientString = message.getStringProperty(MessageEncryptionHelper.ENCRYPT_RECIPIENTS_HEADER); |
| if(recipientString == null) |
| { |
| recipientString = destination.getEncryptedRecipients(); |
| } |
| contentHeaderProperties.getHeaders().remove(MessageEncryptionHelper.ENCRYPT_RECIPIENTS_HEADER); |
| |
| String unencryptedProperties = message.getStringProperty(MessageEncryptionHelper.UNENCRYPTED_PROPERTIES_HEADER); |
| contentHeaderProperties.getHeaders().remove(MessageEncryptionHelper.UNENCRYPTED_PROPERTIES_HEADER); |
| |
| final int headerLength = contentHeaderProperties.getPropertyListSize() + 2; |
| byte[] unencryptedBytes = new byte[headerLength + size]; |
| ByteBuffer output = ByteBuffer.wrap(unencryptedBytes); |
| output.putShort((short) (contentHeaderProperties.getPropertyFlags() & 0xffff)); |
| contentHeaderProperties.writePropertyListPayload(output); |
| |
| if (size != 0) |
| { |
| payload.get(unencryptedBytes, headerLength, payload.remaining()); |
| } |
| |
| byte[] ivbytes = encryptionHelper.getInitialisationVector(); |
| |
| byte[] encryptedBytes = encryptionHelper.encrypt(secretKey, unencryptedBytes, ivbytes); |
| payload = ByteBuffer.wrap(encryptedBytes); |
| |
| if (recipientString == null) |
| { |
| throw new JMSException("When sending an encrypted message, recipients must be supplied"); |
| } |
| String[] recipients = recipientString.split(";"); |
| List<List<Object>> encryptedKeys = new ArrayList<>(); |
| for(MessageEncryptionHelper.KeyTransportRecipientInfo info : encryptionHelper.getKeyTransportRecipientInfo(Arrays.asList(recipients), secretKey)) |
| { |
| encryptedKeys.add(info.asList()); |
| } |
| |
| BasicContentHeaderProperties oldProps = contentHeaderProperties; |
| contentHeaderProperties = new BasicContentHeaderProperties(oldProps); |
| final FieldTable oldHeaders = oldProps.getHeaders(); |
| final FieldTable newHeaders = contentHeaderProperties.getHeaders(); |
| newHeaders.clear(); |
| |
| if(unencryptedProperties != null) |
| { |
| List<String> unencryptedPropertyNames = Arrays.asList(unencryptedProperties.split(" *; *")); |
| for (String propertyName : unencryptedPropertyNames) |
| { |
| if (oldHeaders.propertyExists(propertyName)) |
| { |
| newHeaders.setObject(propertyName, oldHeaders.get(propertyName)); |
| } |
| } |
| } |
| |
| newHeaders.setObject(MessageEncryptionHelper.ENCRYPTED_KEYS_PROPERTY, encryptedKeys); |
| newHeaders.setString(MessageEncryptionHelper.ENCRYPTION_ALGORITHM_PROPERTY, |
| encryptionHelper.getMessageEncryptionCipherName()); |
| newHeaders.setBytes(MessageEncryptionHelper.KEY_INIT_VECTOR_PROPERTY, ivbytes); |
| contentHeaderProperties.setContentType(Encrypted091MessageFactory.ENCRYPTED_0_9_1_CONTENT_TYPE); |
| size = encryptedBytes.length; |
| |
| } |
| catch (GeneralSecurityException | IOException e) |
| { |
| throw JMSExceptionHelper.chainJMSException(new JMSException("Unexpected Exception while encrypting message"), e); |
| } |
| |
| } |
| else |
| { |
| byte[] compressed; |
| if (size > getConnection().getMessageCompressionThresholdSize() |
| && getConnection().getDelegate().isMessageCompressionSupported() |
| && getConnection().isMessageCompressionDesired() |
| && contentHeaderProperties.getEncoding() == null |
| && (compressed = GZIPUtils.compressBufferToArray(payload)) != null) |
| { |
| contentHeaderProperties.setEncoding("gzip"); |
| payload = ByteBuffer.wrap(compressed); |
| size = compressed.length; |
| |
| } |
| } |
| final int contentBodyFrameCount = calculateContentBodyFrameCount(payload); |
| frames = new AMQFrame[2 + contentBodyFrameCount]; |
| |
| if (payload != null) |
| { |
| createContentBodies(payload, frames, 2, getChannelId()); |
| } |
| |
| contentHeaderFrame = |
| ContentHeaderBody.createAMQFrame(getChannelId(), |
| contentHeaderProperties, size); |
| |
| |
| if (getLogger().isDebugEnabled()) |
| { |
| getLogger().debug("Sending " + (frames.length-2) + " content body frames to " + destination); |
| } |
| |
| if (contentHeaderFrame.getSize() > getSession().getAMQConnection().getMaximumFrameSize()) |
| { |
| throw new JMSException("Unable to send message as the headers are too large (" |
| + contentHeaderFrame.getSize() |
| + " bytes, but the maximum negotiated frame size is " |
| + getSession().getAMQConnection().getMaximumFrameSize() |
| + ")."); |
| } |
| if (getLogger().isDebugEnabled()) |
| { |
| getLogger().debug("Sending content header frame to " + destination); |
| } |
| |
| |
| frames[0] = publishFrame; |
| frames[1] = contentHeaderFrame; |
| final CompositeAMQDataBlock compositeFrame = new CompositeAMQDataBlock(frames); |
| |
| try |
| { |
| getSession().checkFlowControl(); |
| } |
| catch (InterruptedException e) |
| { |
| throw JMSExceptionHelper.chainJMSException(new JMSException( |
| "Interrupted while waiting for flow control to be removed"), e); |
| } |
| |
| AMQConnectionDelegate_8_0 connectionDelegate80 = (AMQConnectionDelegate_8_0) (getConnection().getDelegate()); |
| |
| boolean useConfirms = getPublishMode() == PublishMode.SYNC_PUBLISH_ALL |
| && (connectionDelegate80.isConfirmedPublishSupported() |
| || (!getSession().isTransacted() && connectionDelegate80.isConfirmedPublishNonTransactionalSupported())); |
| |
| if(!useConfirms) |
| { |
| getConnection().getProtocolHandler().writeFrame(compositeFrame); |
| } |
| else |
| { |
| final PublishConfirmMessageListener frameListener = new PublishConfirmMessageListener(getChannelId()); |
| try |
| { |
| |
| getConnection().getProtocolHandler().writeCommandFrameAndWaitForReply(compositeFrame, |
| frameListener); |
| |
| if(frameListener.isRejected()) |
| { |
| throw new JMSException("The message was not accepted by the server (e.g. because the address was no longer valid)"); |
| } |
| } |
| catch (QpidException e) |
| { |
| throw JMSExceptionHelper.chainJMSException(new JMSException(e.getMessage()), e); |
| } |
| catch (FailoverException e) |
| { |
| throw JMSExceptionHelper.chainJMSException(new JMSException( |
| "Fail-over interrupted send. Status of the send is uncertain."), e); |
| |
| } |
| } |
| } |
| |
| /** |
| * Create content bodies. This will split a large message into numerous bodies depending on the negotiated |
| * maximum frame size. |
| * |
| * @param payload |
| * @param frames |
| * @param offset |
| * @param channelId @return the array of content bodies |
| */ |
| private void createContentBodies(ByteBuffer payload, AMQFrame[] frames, int offset, int channelId) |
| { |
| |
| if (frames.length == (offset + 1)) |
| { |
| frames[offset] = ContentBody.createAMQFrame(channelId, new ContentBody(payload.slice())); |
| } |
| else |
| { |
| |
| final long framePayloadMax = getMaximumPayloadSize(); |
| long remaining = payload.remaining(); |
| for (int i = offset; i < frames.length; i++) |
| { |
| payload.position((int) framePayloadMax * (i - offset)); |
| int length = (remaining >= framePayloadMax) ? (int) framePayloadMax : (int) remaining; |
| payload.limit(payload.position() + length); |
| |
| frames[i] = ContentBody.createAMQFrame(channelId, new ContentBody(payload.slice())); |
| |
| remaining -= length; |
| } |
| } |
| |
| } |
| |
| private int calculateContentBodyFrameCount(ByteBuffer payload) |
| { |
| // we subtract one from the total frame maximum size to account for the end of frame marker in a body frame |
| // (0xCE byte). |
| int frameCount; |
| if ((payload == null) || (payload.remaining() == 0)) |
| { |
| frameCount = 0; |
| } |
| else |
| { |
| int dataLength = payload.remaining(); |
| final long framePayloadMax = getMaximumPayloadSize(); |
| int lastFrame = ((dataLength % framePayloadMax) > 0) ? 1 : 0; |
| frameCount = (int) (dataLength / framePayloadMax) + lastFrame; |
| } |
| |
| return frameCount; |
| } |
| |
| private long getMaximumPayloadSize() |
| { |
| return getSession().getAMQConnection().getMaximumFrameSize() - 8; |
| } |
| |
| @Override |
| public AMQSession_0_8 getSession() |
| { |
| return (AMQSession_0_8) super.getSession(); |
| } |
| |
| private static class PublishConfirmMessageListener extends BlockingMethodFrameListener |
| { |
| private boolean _rejected; |
| |
| /** |
| * Creates a new method listener, that filters incoming method to just those that match the specified channel id. |
| * |
| * @param channelId The channel id to filter incoming methods with. |
| */ |
| public PublishConfirmMessageListener(final int channelId) |
| { |
| super(channelId); |
| } |
| |
| @Override |
| public boolean processMethod(final int channelId, final AMQMethodBody frame) |
| { |
| if (frame instanceof BasicAckBody) |
| { |
| return true; |
| } |
| else if (frame instanceof BasicNackBody) |
| { |
| _rejected = true; |
| return true; |
| } |
| else |
| { |
| return false; |
| } |
| } |
| |
| public boolean isRejected() |
| { |
| return _rejected; |
| } |
| } |
| } |
