001 /*
002 *
003 * Licensed to the Apache Software Foundation (ASF) under one
004 * or more contributor license agreements. See the NOTICE file
005 * distributed with this work for additional information
006 * regarding copyright ownership. The ASF licenses this file
007 * to you under the Apache License, Version 2.0 (the
008 * "License"); you may not use this file except in compliance
009 * with the License. You may obtain a copy of the License at
010 *
011 * http://www.apache.org/licenses/LICENSE-2.0
012 *
013 * Unless required by applicable law or agreed to in writing,
014 * software distributed under the License is distributed on an
015 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
016 * KIND, either express or implied. See the License for the
017 * specific language governing permissions and limitations
018 * under the License.
019 *
020 */
021 package org.apache.qpid.server.queue;
022
023 import java.util.LinkedList;
024 import java.util.List;
025 import java.util.ArrayList;
026 import java.util.Collections;
027
028 import org.apache.qpid.AMQException;
029 import org.apache.qpid.framing.abstraction.MessagePublishInfo;
030 import org.apache.qpid.framing.BasicContentHeaderProperties;
031 import org.apache.qpid.framing.ContentHeaderBody;
032
033 /**
034 * Contains data that is only used in AMQMessage transiently, e.g. while the content
035 * body fragments are arriving.
036 *
037 * Having this data stored in a separate class means that the AMQMessage class avoids
038 * the small overhead of numerous guaranteed-null references.
039 *
040 * @author Apache Software Foundation
041 */
042 public class TransientMessageData
043 {
044 /**
045 * Stored temporarily until the header has been received at which point it is used when
046 * constructing the handle
047 */
048 private MessagePublishInfo _messagePublishInfo;
049
050 /**
051 * Also stored temporarily.
052 */
053 private ContentHeaderBody _contentHeaderBody;
054
055 /**
056 * Keeps a track of how many bytes we have received in body frames
057 */
058 private long _bodyLengthReceived = 0;
059
060 /**
061 * This is stored during routing, to know the queues to which this message should immediately be
062 * delivered. It is <b>cleared after delivery has been attempted</b>. Any persistent record of destinations is done
063 * by the message handle.
064 */
065 private List<AMQQueue> _destinationQueues;
066
067 public MessagePublishInfo getMessagePublishInfo()
068 {
069 return _messagePublishInfo;
070 }
071
072 public void setMessagePublishInfo(MessagePublishInfo messagePublishInfo)
073 {
074 _messagePublishInfo = messagePublishInfo;
075 }
076
077 public List<AMQQueue> getDestinationQueues()
078 {
079 return _destinationQueues == null ? (List<AMQQueue>) Collections.EMPTY_LIST : _destinationQueues;
080 }
081
082 public void setDestinationQueues(List<AMQQueue> destinationQueues)
083 {
084 _destinationQueues = destinationQueues;
085 }
086
087 public ContentHeaderBody getContentHeaderBody()
088 {
089 return _contentHeaderBody;
090 }
091
092 public void setContentHeaderBody(ContentHeaderBody contentHeaderBody)
093 {
094 _contentHeaderBody = contentHeaderBody;
095 }
096
097 public long getBodyLengthReceived()
098 {
099 return _bodyLengthReceived;
100 }
101
102 public void addBodyLength(int value)
103 {
104 _bodyLengthReceived += value;
105 }
106
107 public boolean isAllContentReceived() throws AMQException
108 {
109 return _bodyLengthReceived == _contentHeaderBody.bodySize;
110 }
111
112 public void addDestinationQueue(AMQQueue queue)
113 {
114 if(_destinationQueues == null)
115 {
116 _destinationQueues = new ArrayList<AMQQueue>();
117 }
118 _destinationQueues.add(queue);
119 }
120
121 public boolean isPersistent()
122 {
123 //todo remove literal values to a constant file such as AMQConstants in common
124 return _contentHeaderBody.properties instanceof BasicContentHeaderProperties &&
125 ((BasicContentHeaderProperties) _contentHeaderBody.properties).getDeliveryMode() == 2;
126 }
127 }
|