You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@activemq.apache.org by RaiSaurabh <gi...@git.apache.org> on 2018/01/19 12:20:31 UTC
[GitHub] activemq-artemis pull request #1793: ARTEMIS-1498: Openwire internal headers...
GitHub user RaiSaurabh opened a pull request:
https://github.com/apache/activemq-artemis/pull/1793
ARTEMIS-1498: Openwire internal headers should not be part of message
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/RaiSaurabh/activemq-artemis openwire
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/activemq-artemis/pull/1793.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #1793
----
commit 29c61ae7ae394ffa955228c98a0ccecc82318d1b
Author: saurabhrai <ra...@...>
Date: 2018-01-19T12:19:05Z
ARTEMIS-1498: Openwire internal headers should not be part of message
----
---
[GitHub] activemq-artemis pull request #1793: ARTEMIS-1498: Openwire internal headers...
Posted by RaiSaurabh <gi...@git.apache.org>.
Github user RaiSaurabh commented on a diff in the pull request:
https://github.com/apache/activemq-artemis/pull/1793#discussion_r168675085
--- Diff: artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireMessage.java ---
@@ -0,0 +1,673 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.activemq.artemis.core.protocol.openwire;
+
+import javax.jms.JMSException;
+import java.io.IOException;
+import java.util.Enumeration;
+import java.util.HashSet;
+import java.util.Set;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
+import org.apache.activemq.artemis.api.core.ActiveMQException;
+import org.apache.activemq.artemis.api.core.ActiveMQPropertyConversionException;
+import org.apache.activemq.artemis.api.core.ICoreMessage;
+import org.apache.activemq.artemis.api.core.Message;
+import org.apache.activemq.artemis.api.core.RefCountMessage;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.buffers.impl.ChannelBufferWrapper;
+import org.apache.activemq.artemis.core.message.impl.CoreMessageObjectPools;
+import org.apache.activemq.artemis.core.persistence.Persister;
+import org.apache.activemq.artemis.utils.DataConstants;
+import org.apache.activemq.command.ActiveMQDestination;
+import org.apache.activemq.command.ActiveMQMessage;
+import org.apache.activemq.openwire.OpenWireFormat;
+import org.apache.activemq.util.ByteSequence;
+import org.apache.activemq.wireformat.WireFormat;
+
+public class OpenWireMessage extends RefCountMessage {
+
+ private org.apache.activemq.command.ActiveMQMessage message;
+ private WireFormat wireFormat;
+ ByteBuf data;
+ boolean bufferValid;
+
+
+ public OpenWireMessage(org.apache.activemq.command.ActiveMQMessage message, WireFormat wireFormat) {
+ this.message = message;
+ this.wireFormat = wireFormat;
+ try {
+ ByteSequence byteSequence = this.wireFormat.marshal(message);
+ setBuffer(Unpooled.copiedBuffer(byteSequence.getData(), byteSequence.getOffset(), byteSequence.getLength()));
+ } catch (IOException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ public OpenWireMessage() {
+ }
+
+ public void setMarsheller(WireFormat wireFormat) {
+ this.wireFormat = wireFormat;
+ }
+
+ public ActiveMQMessage getAMQMessage() {
+ if (message == null) {
+ if (data != null) {
+ try {
+ message = (ActiveMQMessage) wireFormat.unmarshal(new ChannelBufferWrapper(data));
+ } catch (IOException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return message;
+ } else {
+ return new ActiveMQMessage();
+ }
+ }
+ return message;
+ }
+
+ @Override
+ public void messageChanged() {
+ }
--- End diff --
No not at the moment
---
[GitHub] activemq-artemis pull request #1793: ARTEMIS-1498: Openwire internal headers...
Posted by RaiSaurabh <gi...@git.apache.org>.
Github user RaiSaurabh closed the pull request at:
https://github.com/apache/activemq-artemis/pull/1793
---
[GitHub] activemq-artemis issue #1793: ARTEMIS-1498: Openwire internal headers should...
Posted by michaelandrepearce <gi...@git.apache.org>.
Github user michaelandrepearce commented on the issue:
https://github.com/apache/activemq-artemis/pull/1793
Copied from example i gave originally: https://github.com/apache/activemq-artemis/pull/1793
e.g. along this line.
public class OpenwireMessage extends RefCountMessage {
org.apache.activemq.command.Message message;
public OpenwireMessage(org.apache.activemq.command.Message message){
this.message = message;
}
@Override
public SimpleString getReplyTo() {
return SimpleString.toSimpleString(message.getReplyTo().getPhysicalName());
}
@Override
public Message setReplyTo(SimpleString address) {
message.setReplyTo(ActiveMQDestination.createDestination(address.toString(), ActiveMQDestination.QUEUE_TYPE));
return this;
}
@Override
public Object getUserID() {
return message.getUserID();
}
@Override
public Message setUserID(Object userID) {
message.setUserID(userID.toString());
return this;
}
@Override
public boolean isDurable() {
return message.isPersistent();
}
@Override
public Message setDurable(boolean durable) {
message.setPersistent(durable);
return this;
}
.....
}
---
[GitHub] activemq-artemis issue #1793: ARTEMIS-1498: Openwire internal headers should...
Posted by michaelandrepearce <gi...@git.apache.org>.
Github user michaelandrepearce commented on the issue:
https://github.com/apache/activemq-artemis/pull/1793
Here is a much more complete sample, i would be expecting to see (this is what i was meaning in the original PR comment on: #1793)
```
/**
* 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
* <p>
* http://www.apache.org/licenses/LICENSE-2.0
* <p>
* 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.activemq.artemis.core.protocol.openwire;
import javax.jms.JMSException;
import java.io.IOException;
import java.util.Enumeration;
import java.util.HashSet;
import java.util.Set;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufUtil;
import io.netty.buffer.UnpooledByteBufAllocator;
import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
import org.apache.activemq.artemis.api.core.ActiveMQPropertyConversionException;
import org.apache.activemq.artemis.api.core.ICoreMessage;
import org.apache.activemq.artemis.api.core.Message;
import org.apache.activemq.artemis.api.core.RefCountMessage;
import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.core.message.impl.CoreMessageObjectPools;
import org.apache.activemq.artemis.core.persistence.Persister;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.command.ActiveMQMessage;
import org.apache.activemq.util.ByteSequence;
import org.apache.activemq.wireformat.WireFormat;
public class OpenWireMessage extends RefCountMessage {
private org.apache.activemq.command.ActiveMQMessage message;
private WireFormat marshaller;
public OpenWireMessage(org.apache.activemq.command.ActiveMQMessage message, WireFormat marshaller){
this.message = message;
this.marshaller = marshaller;
}
@Override
public void messageChanged() {
}
@Override
public Long getScheduledDeliveryTime() {
return message.getArrival();
}
@Override
public SimpleString getReplyTo() {
return SimpleString.toSimpleString(message.getReplyTo().getPhysicalName());
}
@Override
public Message setReplyTo(SimpleString address) {
message.setReplyTo(ActiveMQDestination.createDestination(address.toString(), ActiveMQDestination.QUEUE_TYPE));
return this;
}
@Override
public Message setBuffer(ByteBuf buffer) {
message.setContent(new ByteSequence(ByteBufUtil.getBytes(buffer)));
return this;
}
@Override
public ByteBuf getBuffer() {
ByteBuf byteBuf = UnpooledByteBufAllocator.DEFAULT.buffer(message.getContent().getLength());
byteBuf.writeBytes(message.getContent().getData(), message.getContent().getOffset(), message.getContent().getLength());
return byteBuf;
}
@Override
public Message copy() {
return new OpenWireMessage((ActiveMQMessage)message.copy(), marshaller);
}
@Override
public Message copy(long newID) {
OpenWireMessage copy = new OpenWireMessage((ActiveMQMessage)message.copy(), marshaller);
copy.setMessageID(newID);
return copy;
}
@Override
public long getMessageID() {
return message.getMessageId().getBrokerSequenceId();
}
@Override
public Message setMessageID(long id) {
message.getMessageId().setBrokerSequenceId(id);
return this;
}
@Override
public long getExpiration() {
return message.getExpiration();
}
@Override
public Message setExpiration(long expiration) {
message.setExpiration(expiration);
return this;
}
@Override
public Object getUserID() {
return message.getUserID();
}
@Override
public Message setUserID(Object userID) {
message.setUserID(userID.toString());
return this;
}
@Override
public boolean isDurable() {
return message.isPersistent();
}
@Override
public Message setDurable(boolean durable) {
message.setPersistent(durable);
return this;
}
@Override
public Persister<Message> getPersister() {
return null;
}
@Override
public String getAddress() {
return message.getDestination().getPhysicalName();
}
@Override
public Message setAddress(String address) {
message.setDestination(ActiveMQDestination.createDestination(address, ActiveMQDestination.QUEUE_TYPE));
return this;
}
@Override
public SimpleString getAddressSimpleString() {
return null;
}
@Override
public Message setAddress(SimpleString address) {
return null;
}
@Override
public long getTimestamp() {
return message.getTimestamp();
}
@Override
public Message setTimestamp(long timestamp) {
message.setTimestamp(timestamp);
return this;
}
@Override
public byte getPriority() {
return message.getPriority();
}
@Override
public Message setPriority(byte priority) {
message.setPriority(priority);
return this;
}
@Override
public void receiveBuffer(ByteBuf buffer) {
}
@Override
public void sendBuffer(ByteBuf buffer, int deliveryCount) {
}
@Override
public int getPersistSize() {
return 0;
}
@Override
public void persist(ActiveMQBuffer targetRecord) {
}
@Override
public void reloadPersistence(ActiveMQBuffer record) {
}
@Override
public Message putBooleanProperty(String key, boolean value) {
try {
message.setBooleanProperty(key, value);
} catch (JMSException e) {
throw new ActiveMQPropertyConversionException(e.getMessage());
}
return this;
}
@Override
public Message putByteProperty(String key, byte value) {
try {
message.setByteProperty(key, value);
} catch (JMSException e) {
throw new ActiveMQPropertyConversionException(e.getMessage());
}
return this;
}
@Override
public Message putBytesProperty(String key, byte[] value) {
return putObjectProperty(key, value);
}
@Override
public Message putShortProperty(String key, short value) {
try {
message.setShortProperty(key, value);
} catch (JMSException e) {
throw new ActiveMQPropertyConversionException(e.getMessage());
}
return this;
}
@Override
public Message putCharProperty(String key, char value) {
return putStringProperty(key, Character.toString(value));
}
@Override
public Message putIntProperty(String key, int value) {
try {
message.setIntProperty(key, value);
} catch (JMSException e) {
throw new ActiveMQPropertyConversionException(e.getMessage());
}
return this;
}
@Override
public Message putLongProperty(String key, long value) {
try {
message.setLongProperty(key, value);
} catch (JMSException e) {
throw new ActiveMQPropertyConversionException(e.getMessage());
}
return this;
}
@Override
public Message putFloatProperty(String key, float value) {
try {
message.setFloatProperty(key, value);
} catch (JMSException e) {
throw new ActiveMQPropertyConversionException(e.getMessage());
}
return this;
}
@Override
public Message putDoubleProperty(String key, double value) {
try {
message.setDoubleProperty(key, value);
} catch (JMSException e) {
throw new ActiveMQPropertyConversionException(e.getMessage());
}
return this;
}
@Override
public Message putBooleanProperty(SimpleString key, boolean value) {
return putBooleanProperty(key.toString(), value);
}
@Override
public Message putByteProperty(SimpleString key, byte value) {
return putByteProperty(key.toString(), value);
}
@Override
public Message putBytesProperty(SimpleString key, byte[] value) {
return putBytesProperty(key.toString(), value);
}
@Override
public Message putShortProperty(SimpleString key, short value) {
return putShortProperty(key.toString(), value);
}
@Override
public Message putCharProperty(SimpleString key, char value) {
return putCharProperty(key.toString(), value);
}
@Override
public Message putIntProperty(SimpleString key, int value) {
return putIntProperty(key.toString(), value);
}
@Override
public Message putLongProperty(SimpleString key, long value) {
return putLongProperty(key.toString(), value);
}
@Override
public Message putFloatProperty(SimpleString key, float value) {
return putFloatProperty(key.toString(), value);
}
@Override
public Message putDoubleProperty(SimpleString key, double value) {
return putDoubleProperty(key.toString(), value);
}
@Override
public Message putStringProperty(String key, String value) {
try {
message.setStringProperty(key, value);
} catch (JMSException e) {
throw new ActiveMQPropertyConversionException(e.getMessage());
}
return this;
}
@Override
public Message putObjectProperty(String key, Object value) throws ActiveMQPropertyConversionException {
try {
final Object v;
if (value instanceof SimpleString) {
v = value.toString();
} else {
v = value;
}
message.setProperty(key, value);
} catch (IOException e) {
throw new ActiveMQPropertyConversionException(e.getMessage());
}
return this;
}
@Override
public Message putObjectProperty(SimpleString key, Object value) throws ActiveMQPropertyConversionException {
return putObjectProperty(key.toString(), value);
}
@Override
public Object removeProperty(String key) {
try {
Object o = message.getProperty(key);
if (o != null) {
message.removeProperty(key);
}
return o;
} catch (IOException e) {
throw new ActiveMQPropertyConversionException(e.getMessage());
}
}
@Override
public boolean containsProperty(String key) {
try {
return message.getProperty(key) != null;
} catch (IOException e) {
throw new ActiveMQPropertyConversionException(e.getMessage());
}
}
@Override
public Boolean getBooleanProperty(String key) throws ActiveMQPropertyConversionException {
try {
return message.getBooleanProperty(key);
} catch (JMSException e) {
throw new ActiveMQPropertyConversionException(e.getMessage());
}
}
@Override
public Byte getByteProperty(String key) throws ActiveMQPropertyConversionException {
try {
return message.getByteProperty(key);
} catch (JMSException e) {
throw new ActiveMQPropertyConversionException(e.getMessage());
}
}
@Override
public Double getDoubleProperty(String key) throws ActiveMQPropertyConversionException {
try {
return message.getDoubleProperty(key);
} catch (JMSException e) {
throw new ActiveMQPropertyConversionException(e.getMessage());
}
}
@Override
public Integer getIntProperty(String key) throws ActiveMQPropertyConversionException {
try {
return message.getIntProperty(key);
} catch (JMSException e) {
throw new ActiveMQPropertyConversionException(e.getMessage());
}
}
@Override
public Long getLongProperty(String key) throws ActiveMQPropertyConversionException {
try {
return message.getLongProperty(key);
} catch (JMSException e) {
throw new ActiveMQPropertyConversionException(e.getMessage());
}
}
@Override
public Object getObjectProperty(String key) {
try {
return message.getObjectProperty(key);
} catch (JMSException e) {
throw new ActiveMQPropertyConversionException(e.getMessage());
}
}
@Override
public Short getShortProperty(String key) throws ActiveMQPropertyConversionException {
try {
return message.getShortProperty(key);
} catch (JMSException e) {
throw new ActiveMQPropertyConversionException(e.getMessage());
}
}
@Override
public Float getFloatProperty(String key) throws ActiveMQPropertyConversionException {
try {
return message.getFloatProperty(key);
} catch (JMSException e) {
throw new ActiveMQPropertyConversionException(e.getMessage());
}
}
@Override
public String getStringProperty(String key) throws ActiveMQPropertyConversionException {
try {
return message.getStringProperty(key);
} catch (JMSException e) {
throw new ActiveMQPropertyConversionException(e.getMessage());
}
}
@Override
public SimpleString getSimpleStringProperty(String key) throws ActiveMQPropertyConversionException {
try {
return SimpleString.toSimpleString(message.getStringProperty(key));
} catch (JMSException e) {
throw new ActiveMQPropertyConversionException(e.getMessage());
}
}
@Override
public byte[] getBytesProperty(String key) throws ActiveMQPropertyConversionException {
try {
return (byte[]) message.getObjectProperty(key);
} catch (JMSException e) {
throw new ActiveMQPropertyConversionException(e.getMessage());
}
}
@Override
public Object removeProperty(SimpleString key) {
return removeProperty(key.toString());
}
@Override
public boolean containsProperty(SimpleString key) {
return containsProperty(key.toString());
}
@Override
public Boolean getBooleanProperty(SimpleString key) throws ActiveMQPropertyConversionException {
return getBooleanProperty(key.toString());
}
@Override
public Byte getByteProperty(SimpleString key) throws ActiveMQPropertyConversionException {
return getByteProperty(key.toString());
}
@Override
public Double getDoubleProperty(SimpleString key) throws ActiveMQPropertyConversionException {
return getDoubleProperty(key.toString());
}
@Override
public Integer getIntProperty(SimpleString key) throws ActiveMQPropertyConversionException {
return getIntProperty(key.toString());
}
@Override
public Long getLongProperty(SimpleString key) throws ActiveMQPropertyConversionException {
return getLongProperty(key.toString());
}
@Override
public Object getObjectProperty(SimpleString key) {
return getObjectProperty(key.toString());
}
@Override
public Object getAnnotation(SimpleString key) {
return getObjectProperty(key.toString());
}
@Override
public Short getShortProperty(SimpleString key) throws ActiveMQPropertyConversionException {
return getShortProperty(key.toString());
}
@Override
public Float getFloatProperty(SimpleString key) throws ActiveMQPropertyConversionException {
return getFloatProperty(key.toString());
}
@Override
public String getStringProperty(SimpleString key) throws ActiveMQPropertyConversionException {
return getStringProperty(key.toString());
}
@Override
public SimpleString getSimpleStringProperty(SimpleString key) throws ActiveMQPropertyConversionException {
return SimpleString.toSimpleString(getStringProperty(key));
}
@Override
public byte[] getBytesProperty(SimpleString key) throws ActiveMQPropertyConversionException {
return getBytesProperty(key.toString());
}
@Override
public Message putStringProperty(SimpleString key, SimpleString value) {
return putStringProperty(key.toString(), value.toString());
}
@Override
public Message putStringProperty(SimpleString key, String value) {
return putStringProperty(key.toString(), value);
}
@Override
public int getEncodeSize() {
return 0;
}
@Override
public Set<SimpleString> getPropertyNames() {
try {
Set<SimpleString> simpleStrings = new HashSet<>();
Enumeration propertyNames = message.getPropertyNames();
while (propertyNames.hasMoreElements()) {
simpleStrings.add(SimpleString.toSimpleString((String) propertyNames.nextElement()));
}
return simpleStrings;
} catch (JMSException e) {
throw new ActiveMQPropertyConversionException(e.getMessage());
}
}
@Override
public ICoreMessage toCore() {
return toCore(null);
}
@Override
public ICoreMessage toCore(CoreMessageObjectPools coreMessageObjectPools) {
try {
return OpenWireMessageConverter.inbound(message, marshaller, coreMessageObjectPools);
} catch (Exception e) {
throw new ActiveMQPropertyConversionException(e.getMessage());
}
}
@Override
public int getMemoryEstimate() {
return message.getSize();
}
}
```
---
[GitHub] activemq-artemis pull request #1793: ARTEMIS-1498: Openwire internal headers...
Posted by michaelandrepearce <gi...@git.apache.org>.
Github user michaelandrepearce commented on a diff in the pull request:
https://github.com/apache/activemq-artemis/pull/1793#discussion_r168166771
--- Diff: artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireMessage.java ---
@@ -0,0 +1,673 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.activemq.artemis.core.protocol.openwire;
+
+import javax.jms.JMSException;
+import java.io.IOException;
+import java.util.Enumeration;
+import java.util.HashSet;
+import java.util.Set;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
+import org.apache.activemq.artemis.api.core.ActiveMQException;
+import org.apache.activemq.artemis.api.core.ActiveMQPropertyConversionException;
+import org.apache.activemq.artemis.api.core.ICoreMessage;
+import org.apache.activemq.artemis.api.core.Message;
+import org.apache.activemq.artemis.api.core.RefCountMessage;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.buffers.impl.ChannelBufferWrapper;
+import org.apache.activemq.artemis.core.message.impl.CoreMessageObjectPools;
+import org.apache.activemq.artemis.core.persistence.Persister;
+import org.apache.activemq.artemis.utils.DataConstants;
+import org.apache.activemq.command.ActiveMQDestination;
+import org.apache.activemq.command.ActiveMQMessage;
+import org.apache.activemq.openwire.OpenWireFormat;
+import org.apache.activemq.util.ByteSequence;
+import org.apache.activemq.wireformat.WireFormat;
+
+public class OpenWireMessage extends RefCountMessage {
+
+ private org.apache.activemq.command.ActiveMQMessage message;
+ private WireFormat wireFormat;
+ ByteBuf data;
+ boolean bufferValid;
+
+
+ public OpenWireMessage(org.apache.activemq.command.ActiveMQMessage message, WireFormat wireFormat) {
+ this.message = message;
+ this.wireFormat = wireFormat;
+ try {
+ ByteSequence byteSequence = this.wireFormat.marshal(message);
+ setBuffer(Unpooled.copiedBuffer(byteSequence.getData(), byteSequence.getOffset(), byteSequence.getLength()));
+ } catch (IOException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ public OpenWireMessage() {
+ }
+
+ public void setMarsheller(WireFormat wireFormat) {
+ this.wireFormat = wireFormat;
+ }
+
+ public ActiveMQMessage getAMQMessage() {
+ if (message == null) {
+ if (data != null) {
+ try {
+ message = (ActiveMQMessage) wireFormat.unmarshal(new ChannelBufferWrapper(data));
+ } catch (IOException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return message;
+ } else {
+ return new ActiveMQMessage();
+ }
+ }
+ return message;
+ }
+
+ @Override
+ public void messageChanged() {
+ }
+
+ @Override
+ public Long getScheduledDeliveryTime() {
+ return message.getArrival();
+ }
+
+ @Override
+ public SimpleString getReplyTo() {
+ return SimpleString.toSimpleString(message.getReplyTo().getPhysicalName());
+ }
+
+ @Override
+ public Message setReplyTo(SimpleString address) {
+ message.setReplyTo(ActiveMQDestination.createDestination(address.toString(), ActiveMQDestination.QUEUE_TYPE));
+ return this;
+ }
+
+ @Override
+ public Message setBuffer(ByteBuf buffer) {
+ this.data = buffer;
+ return this;
+ }
+
+ @Override
+ public ByteBuf getBuffer() {
+ if (data == null) {
+ return null;
+ } else {
+ return Unpooled.wrappedBuffer(data);
+ }
+ }
+
+ @Override
+ public OpenWireMessage copy() {
+ return new OpenWireMessage((ActiveMQMessage)message.copy(), wireFormat);
+ }
+
+ @Override
+ public Message copy(long newID) {
+ OpenWireMessage copy = new OpenWireMessage((ActiveMQMessage)message.copy(), wireFormat);
+ copy.setMessageID(newID);
+ return copy;
+ }
+
+ @Override
+ public long getMessageID() {
+ return message.getMessageId().getBrokerSequenceId();
+ }
+
+ @Override
+ public Message setMessageID(long id) {
+ message.getMessageId().setBrokerSequenceId(id);
+ return this;
+ }
+
+ @Override
+ public long getExpiration() {
+ return message.getExpiration();
+ }
+
+ @Override
+ public Message setExpiration(long expiration) {
+ message.setExpiration(expiration);
+ return this;
+ }
+
+ @Override
+ public Object getUserID() {
+ return message.getUserID();
+ }
+
+ @Override
+ public Message setUserID(Object userID) {
+ message.setUserID(userID.toString());
+ return this;
+ }
+
+ @Override
+ public boolean isDurable() {
+ return message.isPersistent();
+ }
+
+ @Override
+ public Message setDurable(boolean durable) {
+ message.setPersistent(durable);
+ return this;
+ }
+
+ @Override
+ public Persister<Message> getPersister() {
+ return OpenWireMessagePersister.INSTANCE;
+ }
+
+ @Override
+ public String getAddress() {
+ return message.getDestination().getPhysicalName();
+ }
+
+ @Override
+ public Message setAddress(String address) {
+ message.setDestination(ActiveMQDestination.createDestination(address, ActiveMQDestination.QUEUE_TYPE));
+ return this;
+ }
+
+ @Override
+ public SimpleString getAddressSimpleString() {
+ return SimpleString.toSimpleString(message.getDestination().getPhysicalName());
+ }
+
+ @Override
+ public Message setAddress(SimpleString address) {
+ message.setDestination(ActiveMQDestination.createDestination(address.toString(), ActiveMQDestination.QUEUE_TYPE));
+ return this;
+ }
+
+ @Override
+ public long getTimestamp() {
+ return message.getTimestamp();
+ }
+
+ @Override
+ public Message setTimestamp(long timestamp) {
+ message.setTimestamp(timestamp);
+ return this;
+ }
+
+ @Override
+ public byte getPriority() {
+ return message.getPriority();
+ }
+
+ @Override
+ public Message setPriority(byte priority) {
+ message.setPriority(priority);
+ return this;
+ }
+
+ @Override
+ public void receiveBuffer(ByteBuf buffer) {
+
+ }
+
+ @Override
+ public void sendBuffer(ByteBuf buffer, int deliveryCount) {
+
+ }
+
+ @Override
+ public int getPersistSize() {
+ return DataConstants.SIZE_INT + data.array().length;
+ }
+
+ @Override
+ public void persist(ActiveMQBuffer targetRecord) {
+ targetRecord.writeInt(data.writerIndex());
+ targetRecord.writeBytes(data, 0, data.writerIndex());
+ }
+
+ @Override
+ public void reloadPersistence(ActiveMQBuffer record) {
+ int size = record.readInt();
+ byte[] recordArray = new byte[size];
+ record.readBytes(recordArray);
+ this.data = Unpooled.wrappedBuffer(recordArray);
+ OpenWireFormat owf = new OpenWireFormat();
+ try {
+ message = (ActiveMQMessage) owf.unmarshal(new ChannelBufferWrapper(data));
+ } catch (IOException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Message putBooleanProperty(String key, boolean value) {
+ try {
+ message.setBooleanProperty(key, value);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putByteProperty(String key, byte value) {
+ try {
+ message.setByteProperty(key, value);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putBytesProperty(String key, byte[] value) {
+ return putObjectProperty(key, value);
+ }
+
+ @Override
+ public Message putShortProperty(String key, short value) {
+ try {
+ message.setShortProperty(key, value);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putCharProperty(String key, char value) {
+ return putStringProperty(key, Character.toString(value));
+ }
+
+ @Override
+ public Message putIntProperty(String key, int value) {
+ try {
+ message.setIntProperty(key, value);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putLongProperty(String key, long value) {
+ try {
+ message.setLongProperty(key, value);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putFloatProperty(String key, float value) {
+ try {
+ message.setFloatProperty(key, value);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putDoubleProperty(String key, double value) {
+ try {
+ message.setDoubleProperty(key, value);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putBooleanProperty(SimpleString key, boolean value) {
+ return putBooleanProperty(key.toString(), value);
+ }
+
+ @Override
+ public Message putByteProperty(SimpleString key, byte value) {
+ return putByteProperty(key.toString(), value);
+ }
+
+ @Override
+ public Message putBytesProperty(SimpleString key, byte[] value) {
+ return putBytesProperty(key.toString(), value);
+ }
+
+ @Override
+ public Message putShortProperty(SimpleString key, short value) {
+ return putShortProperty(key.toString(), value);
+ }
+
+ @Override
+ public Message putCharProperty(SimpleString key, char value) {
+ return putCharProperty(key.toString(), value);
+ }
+
+ @Override
+ public Message putIntProperty(SimpleString key, int value) {
+ return putIntProperty(key.toString(), value);
+ }
+
+ @Override
+ public Message putLongProperty(SimpleString key, long value) {
+ return putLongProperty(key.toString(), value);
+ }
+
+ @Override
+ public Message putFloatProperty(SimpleString key, float value) {
+ return putFloatProperty(key.toString(), value);
+
+ }
+
+ @Override
+ public Message putDoubleProperty(SimpleString key, double value) {
+ return putDoubleProperty(key.toString(), value);
+ }
+
+ @Override
+ public Message putStringProperty(String key, String value) {
+ try {
+ message.setStringProperty(key, value);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putObjectProperty(String key, Object value) throws ActiveMQPropertyConversionException {
+ try {
+ final Object v;
+ if (value instanceof SimpleString) {
+ v = value.toString();
+ } else {
+ v = value;
+ }
+ message.setProperty(key, value);
+ } catch (IOException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putObjectProperty(SimpleString key, Object value) throws ActiveMQPropertyConversionException {
+ return putObjectProperty(key.toString(), value);
+ }
+
+ @Override
+ public Object removeProperty(String key) {
+ try {
+ Object o = message.getProperty(key);
+ if (o != null) {
+ message.removeProperty(key);
+ }
+ return o;
+ } catch (IOException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+
+ }
+
+ @Override
+ public boolean containsProperty(String key) {
+ try {
+ return message.getProperty(key) != null;
+ } catch (IOException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Boolean getBooleanProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return message.getBooleanProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Byte getByteProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return message.getByteProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Double getDoubleProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return message.getDoubleProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Integer getIntProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return message.getIntProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Long getLongProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return message.getLongProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Object getObjectProperty(String key) {
+ try {
+ return message.getObjectProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Short getShortProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return message.getShortProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Float getFloatProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return message.getFloatProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public String getStringProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return message.getStringProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public SimpleString getSimpleStringProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return SimpleString.toSimpleString(message.getStringProperty(key));
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public byte[] getBytesProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return (byte[]) message.getObjectProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Object removeProperty(SimpleString key) {
+ return removeProperty(key.toString());
+ }
+
+ @Override
+ public boolean containsProperty(SimpleString key) {
+ return containsProperty(key.toString());
+ }
+
+ @Override
+ public Boolean getBooleanProperty(SimpleString key) throws ActiveMQPropertyConversionException {
+ return getBooleanProperty(key.toString());
+ }
+
+ @Override
+ public Byte getByteProperty(SimpleString key) throws ActiveMQPropertyConversionException {
+ return getByteProperty(key.toString());
+ }
+
+ @Override
+ public Double getDoubleProperty(SimpleString key) throws ActiveMQPropertyConversionException {
+ return getDoubleProperty(key.toString());
+ }
+
+ @Override
+ public Integer getIntProperty(SimpleString key) throws ActiveMQPropertyConversionException {
+ return getIntProperty(key.toString());
+ }
+
+ @Override
+ public Long getLongProperty(SimpleString key) throws ActiveMQPropertyConversionException {
+ return getLongProperty(key.toString());
+ }
+
+ @Override
+ public Object getObjectProperty(SimpleString key) {
+ return getObjectProperty(key.toString());
+ }
+
+ @Override
+ public Object getAnnotation(SimpleString key) {
+ return getObjectProperty(key.toString());
+ }
+
+ @Override
+ public Short getShortProperty(SimpleString key) throws ActiveMQPropertyConversionException {
+ return getShortProperty(key.toString());
+ }
+
+ @Override
+ public Float getFloatProperty(SimpleString key) throws ActiveMQPropertyConversionException {
+ return getFloatProperty(key.toString());
+ }
+
+ @Override
+ public String getStringProperty(SimpleString key) throws ActiveMQPropertyConversionException {
+ return getStringProperty(key.toString());
+ }
+
+ @Override
+ public SimpleString getSimpleStringProperty(SimpleString key) throws ActiveMQPropertyConversionException {
+ return SimpleString.toSimpleString(getStringProperty(key));
--- End diff --
Could use pool
---
[GitHub] activemq-artemis issue #1793: ARTEMIS-1498: Openwire internal headers should...
Posted by RaiSaurabh <gi...@git.apache.org>.
Github user RaiSaurabh commented on the issue:
https://github.com/apache/activemq-artemis/pull/1793
@michaelandrepearce @clebertsuconic Apologies for delaying this and dragging this long. I tried to fix the code to pass all the test cases of OpenWire but still, I get 5 failed test cases. I would require your contribution/guidance in fixing the same.
Following are the failed test cases:
OpenWireGroupingTest.testGrouping
JmsTopicRequestReplyTest.testSendAndReceive
RedeliveryPolicyTest.testRedeliveryPolicyPerDestination
TemporaryQueueClusterTest.testClusteredQueue
---
[GitHub] activemq-artemis issue #1793: ARTEMIS-1498: Openwire internal headers should...
Posted by franz1981 <gi...@git.apache.org>.
Github user franz1981 commented on the issue:
https://github.com/apache/activemq-artemis/pull/1793
@RaiSaurabh I'm doing some preliminary benchmarks to see if the number of buffer copies are decreased :+1:
---
[GitHub] activemq-artemis issue #1793: ARTEMIS-1498: Openwire internal headers should...
Posted by franz1981 <gi...@git.apache.org>.
Github user franz1981 commented on the issue:
https://github.com/apache/activemq-artemis/pull/1793
@RaiSaurabh Please rebase it and force push it to re-trigger a Jenkins build and hopefully it will pass
---
[GitHub] activemq-artemis issue #1793: ARTEMIS-1498: Openwire internal headers should...
Posted by michaelandrepearce <gi...@git.apache.org>.
Github user michaelandrepearce commented on the issue:
https://github.com/apache/activemq-artemis/pull/1793
@RaiSaurabh looks really good to me, as noted by @franz1981 there is quite a few bits to optimise, but i personally would prefer we merge this (as long as no functional regressions - which i cannot spot). This way we can all chip in a little on the optimisations over the next week with small little PR's (using the same JIRA)
@franz1981 any objection to this approach?
---
[GitHub] activemq-artemis issue #1793: ARTEMIS-1498: Openwire internal headers should...
Posted by RaiSaurabh <gi...@git.apache.org>.
Github user RaiSaurabh commented on the issue:
https://github.com/apache/activemq-artemis/pull/1793
Ok, @franz1981 I will wait for the merging.
---
[GitHub] activemq-artemis pull request #1793: ARTEMIS-1498: Openwire internal headers...
Posted by michaelandrepearce <gi...@git.apache.org>.
Github user michaelandrepearce commented on a diff in the pull request:
https://github.com/apache/activemq-artemis/pull/1793#discussion_r165768705
--- Diff: artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireCoreConverter.java ---
@@ -0,0 +1,572 @@
+/*
+ * 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.activemq.artemis.core.protocol.openwire;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Arrays;
+import java.util.Map;
+import java.util.Set;
+import java.util.Map.Entry;
+import java.util.zip.Deflater;
+import java.util.zip.DeflaterOutputStream;
+import java.util.zip.Inflater;
+import java.util.zip.InflaterInputStream;
+import java.util.zip.InflaterOutputStream;
+
+import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
+import org.apache.activemq.artemis.api.core.FilterConstants;
+import org.apache.activemq.artemis.api.core.ICoreMessage;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.message.impl.CoreMessage;
+import org.apache.activemq.artemis.core.message.impl.CoreMessageObjectPools;
+import org.apache.activemq.artemis.reader.MessageUtil;
+import org.apache.activemq.artemis.utils.DataConstants;
+import org.apache.activemq.artemis.utils.collections.TypedProperties;
+import org.apache.activemq.command.CommandTypes;
+import org.apache.activemq.util.ByteArrayInputStream;
+import org.apache.activemq.util.ByteSequence;
+import org.apache.activemq.util.ByteSequenceData;
+import org.apache.activemq.util.MarshallingSupport;
+import org.fusesource.hawtbuf.UTF8Buffer;
+
+public class OpenWireCoreConverter {
+ public static final SimpleString JMS_GROUPSEQ = new SimpleString("JMSXGroupSequence");
+ public static final SimpleString JMS_XGROUPSEQ = new SimpleString("JMSXGroupSeq");
+ public static final SimpleString JMS_GROUPID = new SimpleString("JMSXGroupID");
+ private static final SimpleString AMQP_REPLYTOGROUPID = new SimpleString("JMS_AMQP_ReplyToGroupID");
+ public static final SimpleString JMS_CONTENT_TYPE = new SimpleString("JMS_AMQP_ContentType");
+ private static final SimpleString AMQP_CONTETTYPE = new SimpleString("JMS_AMQP_CONTENT_TYPE");
+
+ public static ICoreMessage toCore(final OpenWireMessage openwireMessage, final CoreMessageObjectPools coreMessageObjectPools)
+ throws Exception {
+ CoreMessage coreMessage = new CoreMessage(-1, openwireMessage.getMessageSize(), coreMessageObjectPools);
+ final SimpleString type = openwireMessage.getSimpleStringProperty(OpenWireMessageConverter.JMS_TYPE_PROPERTY);
+ if (type != null) {
+ coreMessage.putStringProperty(OpenWireMessageConverter.JMS_TYPE_PROPERTY, type);
+ }
+ coreMessage.setDurable(openwireMessage.isDurable());
+ coreMessage.setExpiration(openwireMessage.getExpiration());
+ coreMessage.setPriority(openwireMessage.getPriority());
+ coreMessage.setTimestamp(openwireMessage.getTimestamp());
+ final ActiveMQBuffer openWireMessageBuffer = openwireMessage.getReadOnlyBodyBuffer();
+ final boolean isCompressed = openwireMessage.getBooleanProperty(OpenWireMessageConverter.AMQ_MSG_COMPRESSED);
+ final byte coreType = openwireMessage.getType();
+ coreMessage.setType(coreType);
+ openWireMessageBuffer.resetReaderIndex();
+ final byte[] bytes;
+ bytes = readContentFromBody(openWireMessageBuffer, coreType, isCompressed);
+ if (bytes != null) {
+ ByteSequence content = new ByteSequence(bytes);
+ ActiveMQBuffer coreMessageBuffer = coreMessage.getBodyBuffer();
+ writeContentIntoBody(coreMessageBuffer, content, coreType, isCompressed);
+ }
+ coreMessage.putObjectProperty(FilterConstants.NATIVE_MESSAGE_ID,
+ openwireMessage.getObjectProperty(FilterConstants.NATIVE_MESSAGE_ID));
+ coreMessage.setMessageID(openwireMessage.getMessageID());
+ final String corrId = openwireMessage.getStringProperty(OpenWireMessageConverter.JMS_CORRELATION_ID_PROPERTY);
+ if (corrId != null) {
+ coreMessage.putStringProperty(OpenWireMessageConverter.JMS_CORRELATION_ID_PROPERTY, corrId);
+ }
+ final String groupId = openwireMessage.getStringProperty(org.apache.activemq.artemis.api.core.Message.HDR_GROUP_ID);
+ if (groupId != null) {
+ coreMessage.putStringProperty(org.apache.activemq.artemis.api.core.Message.HDR_GROUP_ID,
+ new SimpleString(groupId));
+ coreMessage.putStringProperty(JMS_GROUPID, new SimpleString(groupId));
+ }
+ if (openwireMessage.containsProperty(JMS_GROUPSEQ)) {
+ coreMessage.putIntProperty(JMS_GROUPSEQ, openwireMessage.getIntProperty(JMS_GROUPSEQ));
+ coreMessage.putIntProperty(JMS_XGROUPSEQ, openwireMessage.getIntProperty(JMS_GROUPSEQ));
+ }
+ final SimpleString lastValueProperty = openwireMessage.getLastValueProperty();
+ if (lastValueProperty != null) {
+ coreMessage.setLastValueProperty(lastValueProperty);
+ }
+ coreMessage.putObjectProperty(AMQP_CONTETTYPE, openwireMessage.getObjectProperty(JMS_CONTENT_TYPE));
+ coreMessage.putObjectProperty(AMQP_REPLYTOGROUPID, openwireMessage.getObjectProperty(AMQP_REPLYTOGROUPID));
+ coreMessage.setReplyTo(openwireMessage.getReplyTo());
+ coreMessage.setUserID(openwireMessage.getUserID());
+ coreMessage.setDurable(openwireMessage.isDurable());
+ coreMessage.setAddress(openwireMessage.getAddress());
+ coreMessage.putObjectProperty(org.apache.activemq.artemis.api.core.Message.HDR_GROUP_ID,
+ (SimpleString) openwireMessage
+ .getObjectProperty(org.apache.activemq.artemis.api.core.Message.HDR_GROUP_ID));
+ copyAllProperties(openwireMessage, coreMessage);
+ return coreMessage;
+ }
+
+ public static byte toCoreType(byte amqType) {
+ switch (amqType) {
+ case CommandTypes.ACTIVEMQ_BLOB_MESSAGE:
+ throw new IllegalStateException("We don't support BLOB type yet!");
+ case CommandTypes.ACTIVEMQ_BYTES_MESSAGE:
+ return org.apache.activemq.artemis.api.core.Message.BYTES_TYPE;
+ case CommandTypes.ACTIVEMQ_MAP_MESSAGE:
+ return org.apache.activemq.artemis.api.core.Message.MAP_TYPE;
+ case CommandTypes.ACTIVEMQ_OBJECT_MESSAGE:
+ return org.apache.activemq.artemis.api.core.Message.OBJECT_TYPE;
+ case CommandTypes.ACTIVEMQ_STREAM_MESSAGE:
+ return org.apache.activemq.artemis.api.core.Message.STREAM_TYPE;
+ case CommandTypes.ACTIVEMQ_TEXT_MESSAGE:
+ return org.apache.activemq.artemis.api.core.Message.TEXT_TYPE;
+ case CommandTypes.ACTIVEMQ_MESSAGE:
+ return org.apache.activemq.artemis.api.core.Message.DEFAULT_TYPE;
+ default:
+ throw new IllegalStateException("Unknown ActiveMQ Artemis message type: " + amqType);
+ }
+ }
+
+ public static void loadMapIntoProperties(TypedProperties props, Map<String, Object> map) {
+ for (Entry<String, Object> entry : map.entrySet()) {
+ SimpleString key = new SimpleString(entry.getKey());
+ Object value = entry.getValue();
+ if (value instanceof UTF8Buffer) {
+ value = ((UTF8Buffer) value).toString();
--- End diff --
@franz1981 as per other comment, if you're not dead against it, i personally merge this PR, and then we (me and you) can go through with some TLC on it.
---
[GitHub] activemq-artemis issue #1793: ARTEMIS-1498: Openwire internal headers should...
Posted by RaiSaurabh <gi...@git.apache.org>.
Github user RaiSaurabh commented on the issue:
https://github.com/apache/activemq-artemis/pull/1793
@michaelandrepearce Looks like another issue with GIT repo. Could you please check.
---
[GitHub] activemq-artemis issue #1793: ARTEMIS-1498: Openwire internal headers should...
Posted by RaiSaurabh <gi...@git.apache.org>.
Github user RaiSaurabh commented on the issue:
https://github.com/apache/activemq-artemis/pull/1793
Ok @michaelandrepearce .
---
[GitHub] activemq-artemis issue #1793: ARTEMIS-1498: Openwire internal headers should...
Posted by RaiSaurabh <gi...@git.apache.org>.
Github user RaiSaurabh commented on the issue:
https://github.com/apache/activemq-artemis/pull/1793
@michaelandrepearce Seems there is some problem with the Git due to which this build failed. Could you please check and let me know.
---
[GitHub] activemq-artemis pull request #1793: ARTEMIS-1498: Openwire internal headers...
Posted by michaelandrepearce <gi...@git.apache.org>.
Github user michaelandrepearce commented on a diff in the pull request:
https://github.com/apache/activemq-artemis/pull/1793#discussion_r168166093
--- Diff: artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireMessage.java ---
@@ -0,0 +1,673 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.activemq.artemis.core.protocol.openwire;
+
+import javax.jms.JMSException;
+import java.io.IOException;
+import java.util.Enumeration;
+import java.util.HashSet;
+import java.util.Set;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
+import org.apache.activemq.artemis.api.core.ActiveMQException;
+import org.apache.activemq.artemis.api.core.ActiveMQPropertyConversionException;
+import org.apache.activemq.artemis.api.core.ICoreMessage;
+import org.apache.activemq.artemis.api.core.Message;
+import org.apache.activemq.artemis.api.core.RefCountMessage;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.buffers.impl.ChannelBufferWrapper;
+import org.apache.activemq.artemis.core.message.impl.CoreMessageObjectPools;
+import org.apache.activemq.artemis.core.persistence.Persister;
+import org.apache.activemq.artemis.utils.DataConstants;
+import org.apache.activemq.command.ActiveMQDestination;
+import org.apache.activemq.command.ActiveMQMessage;
+import org.apache.activemq.openwire.OpenWireFormat;
+import org.apache.activemq.util.ByteSequence;
+import org.apache.activemq.wireformat.WireFormat;
+
+public class OpenWireMessage extends RefCountMessage {
+
+ private org.apache.activemq.command.ActiveMQMessage message;
+ private WireFormat wireFormat;
+ ByteBuf data;
+ boolean bufferValid;
+
+
+ public OpenWireMessage(org.apache.activemq.command.ActiveMQMessage message, WireFormat wireFormat) {
+ this.message = message;
+ this.wireFormat = wireFormat;
+ try {
+ ByteSequence byteSequence = this.wireFormat.marshal(message);
+ setBuffer(Unpooled.copiedBuffer(byteSequence.getData(), byteSequence.getOffset(), byteSequence.getLength()));
+ } catch (IOException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ public OpenWireMessage() {
+ }
+
+ public void setMarsheller(WireFormat wireFormat) {
+ this.wireFormat = wireFormat;
+ }
+
+ public ActiveMQMessage getAMQMessage() {
+ if (message == null) {
+ if (data != null) {
+ try {
+ message = (ActiveMQMessage) wireFormat.unmarshal(new ChannelBufferWrapper(data));
+ } catch (IOException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return message;
+ } else {
+ return new ActiveMQMessage();
+ }
+ }
+ return message;
+ }
+
+ @Override
+ public void messageChanged() {
+ }
+
+ @Override
+ public Long getScheduledDeliveryTime() {
+ return message.getArrival();
+ }
+
+ @Override
+ public SimpleString getReplyTo() {
+ return SimpleString.toSimpleString(message.getReplyTo().getPhysicalName());
+ }
+
+ @Override
+ public Message setReplyTo(SimpleString address) {
+ message.setReplyTo(ActiveMQDestination.createDestination(address.toString(), ActiveMQDestination.QUEUE_TYPE));
+ return this;
+ }
+
+ @Override
+ public Message setBuffer(ByteBuf buffer) {
+ this.data = buffer;
+ return this;
+ }
+
+ @Override
+ public ByteBuf getBuffer() {
+ if (data == null) {
+ return null;
+ } else {
+ return Unpooled.wrappedBuffer(data);
+ }
+ }
+
+ @Override
+ public OpenWireMessage copy() {
+ return new OpenWireMessage((ActiveMQMessage)message.copy(), wireFormat);
+ }
+
+ @Override
+ public Message copy(long newID) {
+ OpenWireMessage copy = new OpenWireMessage((ActiveMQMessage)message.copy(), wireFormat);
+ copy.setMessageID(newID);
+ return copy;
+ }
+
+ @Override
+ public long getMessageID() {
+ return message.getMessageId().getBrokerSequenceId();
+ }
+
+ @Override
+ public Message setMessageID(long id) {
+ message.getMessageId().setBrokerSequenceId(id);
+ return this;
+ }
+
+ @Override
+ public long getExpiration() {
+ return message.getExpiration();
+ }
+
+ @Override
+ public Message setExpiration(long expiration) {
+ message.setExpiration(expiration);
+ return this;
+ }
+
+ @Override
+ public Object getUserID() {
+ return message.getUserID();
+ }
+
+ @Override
+ public Message setUserID(Object userID) {
+ message.setUserID(userID.toString());
+ return this;
+ }
+
+ @Override
+ public boolean isDurable() {
+ return message.isPersistent();
+ }
+
+ @Override
+ public Message setDurable(boolean durable) {
+ message.setPersistent(durable);
+ return this;
+ }
+
+ @Override
+ public Persister<Message> getPersister() {
+ return OpenWireMessagePersister.INSTANCE;
+ }
+
+ @Override
+ public String getAddress() {
+ return message.getDestination().getPhysicalName();
+ }
+
+ @Override
+ public Message setAddress(String address) {
+ message.setDestination(ActiveMQDestination.createDestination(address, ActiveMQDestination.QUEUE_TYPE));
+ return this;
+ }
+
+ @Override
+ public SimpleString getAddressSimpleString() {
+ return SimpleString.toSimpleString(message.getDestination().getPhysicalName());
+ }
+
+ @Override
+ public Message setAddress(SimpleString address) {
+ message.setDestination(ActiveMQDestination.createDestination(address.toString(), ActiveMQDestination.QUEUE_TYPE));
+ return this;
+ }
+
+ @Override
+ public long getTimestamp() {
+ return message.getTimestamp();
+ }
+
+ @Override
+ public Message setTimestamp(long timestamp) {
+ message.setTimestamp(timestamp);
+ return this;
+ }
+
+ @Override
+ public byte getPriority() {
+ return message.getPriority();
+ }
+
+ @Override
+ public Message setPriority(byte priority) {
+ message.setPriority(priority);
+ return this;
+ }
+
+ @Override
+ public void receiveBuffer(ByteBuf buffer) {
+
+ }
+
+ @Override
+ public void sendBuffer(ByteBuf buffer, int deliveryCount) {
--- End diff --
Needs implementing?
---
[GitHub] activemq-artemis pull request #1793: ARTEMIS-1498: Openwire internal headers...
Posted by RaiSaurabh <gi...@git.apache.org>.
Github user RaiSaurabh commented on a diff in the pull request:
https://github.com/apache/activemq-artemis/pull/1793#discussion_r168675067
--- Diff: artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireMessage.java ---
@@ -0,0 +1,673 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.activemq.artemis.core.protocol.openwire;
+
+import javax.jms.JMSException;
+import java.io.IOException;
+import java.util.Enumeration;
+import java.util.HashSet;
+import java.util.Set;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
+import org.apache.activemq.artemis.api.core.ActiveMQException;
+import org.apache.activemq.artemis.api.core.ActiveMQPropertyConversionException;
+import org.apache.activemq.artemis.api.core.ICoreMessage;
+import org.apache.activemq.artemis.api.core.Message;
+import org.apache.activemq.artemis.api.core.RefCountMessage;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.buffers.impl.ChannelBufferWrapper;
+import org.apache.activemq.artemis.core.message.impl.CoreMessageObjectPools;
+import org.apache.activemq.artemis.core.persistence.Persister;
+import org.apache.activemq.artemis.utils.DataConstants;
+import org.apache.activemq.command.ActiveMQDestination;
+import org.apache.activemq.command.ActiveMQMessage;
+import org.apache.activemq.openwire.OpenWireFormat;
+import org.apache.activemq.util.ByteSequence;
+import org.apache.activemq.wireformat.WireFormat;
+
+public class OpenWireMessage extends RefCountMessage {
+
+ private org.apache.activemq.command.ActiveMQMessage message;
+ private WireFormat wireFormat;
+ ByteBuf data;
+ boolean bufferValid;
+
+
+ public OpenWireMessage(org.apache.activemq.command.ActiveMQMessage message, WireFormat wireFormat) {
+ this.message = message;
+ this.wireFormat = wireFormat;
+ try {
+ ByteSequence byteSequence = this.wireFormat.marshal(message);
+ setBuffer(Unpooled.copiedBuffer(byteSequence.getData(), byteSequence.getOffset(), byteSequence.getLength()));
+ } catch (IOException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ public OpenWireMessage() {
+ }
+
+ public void setMarsheller(WireFormat wireFormat) {
+ this.wireFormat = wireFormat;
+ }
+
+ public ActiveMQMessage getAMQMessage() {
+ if (message == null) {
+ if (data != null) {
+ try {
+ message = (ActiveMQMessage) wireFormat.unmarshal(new ChannelBufferWrapper(data));
+ } catch (IOException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return message;
+ } else {
+ return new ActiveMQMessage();
+ }
+ }
+ return message;
+ }
+
+ @Override
+ public void messageChanged() {
+ }
+
+ @Override
+ public Long getScheduledDeliveryTime() {
+ return message.getArrival();
+ }
+
+ @Override
+ public SimpleString getReplyTo() {
+ return SimpleString.toSimpleString(message.getReplyTo().getPhysicalName());
+ }
+
+ @Override
+ public Message setReplyTo(SimpleString address) {
+ message.setReplyTo(ActiveMQDestination.createDestination(address.toString(), ActiveMQDestination.QUEUE_TYPE));
+ return this;
+ }
+
+ @Override
+ public Message setBuffer(ByteBuf buffer) {
+ this.data = buffer;
+ return this;
+ }
+
+ @Override
+ public ByteBuf getBuffer() {
+ if (data == null) {
+ return null;
+ } else {
+ return Unpooled.wrappedBuffer(data);
+ }
+ }
+
+ @Override
+ public OpenWireMessage copy() {
+ return new OpenWireMessage((ActiveMQMessage)message.copy(), wireFormat);
+ }
+
+ @Override
+ public Message copy(long newID) {
+ OpenWireMessage copy = new OpenWireMessage((ActiveMQMessage)message.copy(), wireFormat);
+ copy.setMessageID(newID);
+ return copy;
+ }
+
+ @Override
+ public long getMessageID() {
+ return message.getMessageId().getBrokerSequenceId();
+ }
+
+ @Override
+ public Message setMessageID(long id) {
+ message.getMessageId().setBrokerSequenceId(id);
+ return this;
+ }
+
+ @Override
+ public long getExpiration() {
+ return message.getExpiration();
+ }
+
+ @Override
+ public Message setExpiration(long expiration) {
+ message.setExpiration(expiration);
+ return this;
+ }
+
+ @Override
+ public Object getUserID() {
+ return message.getUserID();
+ }
+
+ @Override
+ public Message setUserID(Object userID) {
+ message.setUserID(userID.toString());
+ return this;
+ }
+
+ @Override
+ public boolean isDurable() {
+ return message.isPersistent();
+ }
+
+ @Override
+ public Message setDurable(boolean durable) {
+ message.setPersistent(durable);
+ return this;
+ }
+
+ @Override
+ public Persister<Message> getPersister() {
+ return OpenWireMessagePersister.INSTANCE;
+ }
+
+ @Override
+ public String getAddress() {
+ return message.getDestination().getPhysicalName();
+ }
+
+ @Override
+ public Message setAddress(String address) {
+ message.setDestination(ActiveMQDestination.createDestination(address, ActiveMQDestination.QUEUE_TYPE));
+ return this;
+ }
+
+ @Override
+ public SimpleString getAddressSimpleString() {
+ return SimpleString.toSimpleString(message.getDestination().getPhysicalName());
+ }
+
+ @Override
+ public Message setAddress(SimpleString address) {
+ message.setDestination(ActiveMQDestination.createDestination(address.toString(), ActiveMQDestination.QUEUE_TYPE));
+ return this;
+ }
+
+ @Override
+ public long getTimestamp() {
+ return message.getTimestamp();
+ }
+
+ @Override
+ public Message setTimestamp(long timestamp) {
+ message.setTimestamp(timestamp);
+ return this;
+ }
+
+ @Override
+ public byte getPriority() {
+ return message.getPriority();
+ }
+
+ @Override
+ public Message setPriority(byte priority) {
+ message.setPriority(priority);
+ return this;
+ }
+
+ @Override
+ public void receiveBuffer(ByteBuf buffer) {
+
+ }
+
+ @Override
+ public void sendBuffer(ByteBuf buffer, int deliveryCount) {
+
+ }
+
+ @Override
+ public int getPersistSize() {
+ return DataConstants.SIZE_INT + data.array().length;
+ }
+
+ @Override
+ public void persist(ActiveMQBuffer targetRecord) {
+ targetRecord.writeInt(data.writerIndex());
+ targetRecord.writeBytes(data, 0, data.writerIndex());
+ }
+
+ @Override
+ public void reloadPersistence(ActiveMQBuffer record) {
+ int size = record.readInt();
+ byte[] recordArray = new byte[size];
+ record.readBytes(recordArray);
+ this.data = Unpooled.wrappedBuffer(recordArray);
+ OpenWireFormat owf = new OpenWireFormat();
+ try {
+ message = (ActiveMQMessage) owf.unmarshal(new ChannelBufferWrapper(data));
+ } catch (IOException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Message putBooleanProperty(String key, boolean value) {
+ try {
+ message.setBooleanProperty(key, value);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putByteProperty(String key, byte value) {
+ try {
+ message.setByteProperty(key, value);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putBytesProperty(String key, byte[] value) {
+ return putObjectProperty(key, value);
+ }
+
+ @Override
+ public Message putShortProperty(String key, short value) {
+ try {
+ message.setShortProperty(key, value);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putCharProperty(String key, char value) {
+ return putStringProperty(key, Character.toString(value));
+ }
+
+ @Override
+ public Message putIntProperty(String key, int value) {
+ try {
+ message.setIntProperty(key, value);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putLongProperty(String key, long value) {
+ try {
+ message.setLongProperty(key, value);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putFloatProperty(String key, float value) {
+ try {
+ message.setFloatProperty(key, value);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putDoubleProperty(String key, double value) {
+ try {
+ message.setDoubleProperty(key, value);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putBooleanProperty(SimpleString key, boolean value) {
+ return putBooleanProperty(key.toString(), value);
+ }
+
+ @Override
+ public Message putByteProperty(SimpleString key, byte value) {
+ return putByteProperty(key.toString(), value);
+ }
+
+ @Override
+ public Message putBytesProperty(SimpleString key, byte[] value) {
+ return putBytesProperty(key.toString(), value);
+ }
+
+ @Override
+ public Message putShortProperty(SimpleString key, short value) {
+ return putShortProperty(key.toString(), value);
+ }
+
+ @Override
+ public Message putCharProperty(SimpleString key, char value) {
+ return putCharProperty(key.toString(), value);
+ }
+
+ @Override
+ public Message putIntProperty(SimpleString key, int value) {
+ return putIntProperty(key.toString(), value);
+ }
+
+ @Override
+ public Message putLongProperty(SimpleString key, long value) {
+ return putLongProperty(key.toString(), value);
+ }
+
+ @Override
+ public Message putFloatProperty(SimpleString key, float value) {
+ return putFloatProperty(key.toString(), value);
+
+ }
+
+ @Override
+ public Message putDoubleProperty(SimpleString key, double value) {
+ return putDoubleProperty(key.toString(), value);
+ }
+
+ @Override
+ public Message putStringProperty(String key, String value) {
+ try {
+ message.setStringProperty(key, value);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putObjectProperty(String key, Object value) throws ActiveMQPropertyConversionException {
+ try {
+ final Object v;
+ if (value instanceof SimpleString) {
+ v = value.toString();
+ } else {
+ v = value;
+ }
+ message.setProperty(key, value);
+ } catch (IOException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putObjectProperty(SimpleString key, Object value) throws ActiveMQPropertyConversionException {
+ return putObjectProperty(key.toString(), value);
+ }
+
+ @Override
+ public Object removeProperty(String key) {
+ try {
+ Object o = message.getProperty(key);
+ if (o != null) {
+ message.removeProperty(key);
+ }
+ return o;
+ } catch (IOException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+
+ }
+
+ @Override
+ public boolean containsProperty(String key) {
+ try {
+ return message.getProperty(key) != null;
+ } catch (IOException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Boolean getBooleanProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return message.getBooleanProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Byte getByteProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return message.getByteProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Double getDoubleProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return message.getDoubleProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Integer getIntProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return message.getIntProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Long getLongProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return message.getLongProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Object getObjectProperty(String key) {
+ try {
+ return message.getObjectProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Short getShortProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return message.getShortProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Float getFloatProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return message.getFloatProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public String getStringProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return message.getStringProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public SimpleString getSimpleStringProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return SimpleString.toSimpleString(message.getStringProperty(key));
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public byte[] getBytesProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return (byte[]) message.getObjectProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Object removeProperty(SimpleString key) {
+ return removeProperty(key.toString());
+ }
+
+ @Override
+ public boolean containsProperty(SimpleString key) {
+ return containsProperty(key.toString());
+ }
+
+ @Override
+ public Boolean getBooleanProperty(SimpleString key) throws ActiveMQPropertyConversionException {
+ return getBooleanProperty(key.toString());
+ }
+
+ @Override
+ public Byte getByteProperty(SimpleString key) throws ActiveMQPropertyConversionException {
+ return getByteProperty(key.toString());
+ }
+
+ @Override
+ public Double getDoubleProperty(SimpleString key) throws ActiveMQPropertyConversionException {
+ return getDoubleProperty(key.toString());
+ }
+
+ @Override
+ public Integer getIntProperty(SimpleString key) throws ActiveMQPropertyConversionException {
+ return getIntProperty(key.toString());
+ }
+
+ @Override
+ public Long getLongProperty(SimpleString key) throws ActiveMQPropertyConversionException {
+ return getLongProperty(key.toString());
+ }
+
+ @Override
+ public Object getObjectProperty(SimpleString key) {
+ return getObjectProperty(key.toString());
+ }
+
+ @Override
+ public Object getAnnotation(SimpleString key) {
+ return getObjectProperty(key.toString());
+ }
+
+ @Override
+ public Short getShortProperty(SimpleString key) throws ActiveMQPropertyConversionException {
+ return getShortProperty(key.toString());
+ }
+
+ @Override
+ public Float getFloatProperty(SimpleString key) throws ActiveMQPropertyConversionException {
+ return getFloatProperty(key.toString());
+ }
+
+ @Override
+ public String getStringProperty(SimpleString key) throws ActiveMQPropertyConversionException {
+ return getStringProperty(key.toString());
+ }
+
+ @Override
+ public SimpleString getSimpleStringProperty(SimpleString key) throws ActiveMQPropertyConversionException {
+ return SimpleString.toSimpleString(getStringProperty(key));
+ }
+
+ @Override
+ public byte[] getBytesProperty(SimpleString key) throws ActiveMQPropertyConversionException {
+ return getBytesProperty(key.toString());
+ }
+
+ @Override
+ public Message putStringProperty(SimpleString key, SimpleString value) {
+ return putStringProperty(key.toString(), value.toString());
+ }
+
+ @Override
+ public Message putStringProperty(SimpleString key, String value) {
+ return putStringProperty(key.toString(), value);
+ }
+
+ @Override
+ public int getEncodeSize() {
+ return 0;
+ }
+
+ @Override
+ public Set<SimpleString> getPropertyNames() {
+ try {
+ Set<SimpleString> simpleStrings = new HashSet<>();
+ Enumeration propertyNames = message.getPropertyNames();
+ while (propertyNames.hasMoreElements()) {
+ simpleStrings.add(SimpleString.toSimpleString((String) propertyNames.nextElement()));
+ }
+ return simpleStrings;
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public ICoreMessage toCore() {
+ return toCore(null);
+ }
+
+ @Override
+ public ICoreMessage toCore(CoreMessageObjectPools coreMessageObjectPools) {
+ try {
+ return OpenWireMessageConverter.convertAMQToCore(message, coreMessageObjectPools);
+ } catch (Exception e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public int getMemoryEstimate() {
+ return message.getSize();
+ }
+
+ @Override
+ public Object getDuplicateProperty() {
+ return getObjectProperty(Message.HDR_DUPLICATE_DETECTION_ID);
+ }
+
+ @Override
+ public SimpleString getLastValueProperty() {
+ return getSimpleStringProperty(Message.HDR_LAST_VALUE_NAME);
--- End diff --
Done
---
[GitHub] activemq-artemis pull request #1793: ARTEMIS-1498: Openwire internal headers...
Posted by franz1981 <gi...@git.apache.org>.
Github user franz1981 commented on a diff in the pull request:
https://github.com/apache/activemq-artemis/pull/1793#discussion_r165609280
--- Diff: artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireMessagePersister.java ---
@@ -0,0 +1,64 @@
+/*
+ * 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.activemq.artemis.core.protocol.openwire;
+
+import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
+import org.apache.activemq.artemis.api.core.Message;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.persistence.Persister;
+import org.apache.activemq.artemis.utils.DataConstants;
+
+public enum OpenWireMessagePersister implements Persister<Message> {
+
+ INSTANCE;
+ public static final byte ID = 2;
+
+ public static OpenWireMessagePersister getInstance() {
--- End diff --
optional: this method is not neede anymore
---
[GitHub] activemq-artemis issue #1793: ARTEMIS-1498: Openwire internal headers should...
Posted by michaelandrepearce <gi...@git.apache.org>.
Github user michaelandrepearce commented on the issue:
https://github.com/apache/activemq-artemis/pull/1793
@clebertsuconic you able to give @raisaurabh a pr with what you mean. I’ve gone though it myself and in general looks fine from what I can tell, eg it has a check to see if it’s of openwire message type and if it is it returns it instead of converting
---
[GitHub] activemq-artemis pull request #1793: ARTEMIS-1498: Openwire internal headers...
Posted by franz1981 <gi...@git.apache.org>.
Github user franz1981 commented on a diff in the pull request:
https://github.com/apache/activemq-artemis/pull/1793#discussion_r165604393
--- Diff: artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireMessage.java ---
@@ -0,0 +1,1152 @@
+/**
+ * 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.activemq.artemis.core.protocol.openwire;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.util.LinkedList;
+import java.util.Set;
+
+import io.netty.buffer.ByteBuf;
+import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
+import org.apache.activemq.artemis.api.core.ActiveMQBuffers;
+import org.apache.activemq.artemis.api.core.ActiveMQException;
+import org.apache.activemq.artemis.api.core.ActiveMQPropertyConversionException;
+import org.apache.activemq.artemis.api.core.ICoreMessage;
+import org.apache.activemq.artemis.api.core.Message;
+import org.apache.activemq.artemis.api.core.RefCountMessage;
+import org.apache.activemq.artemis.api.core.RoutingType;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.buffers.impl.ChannelBufferWrapper;
+import org.apache.activemq.artemis.core.buffers.impl.ResetLimitWrappedActiveMQBuffer;
+import org.apache.activemq.artemis.core.message.LargeBodyEncoder;
+import org.apache.activemq.artemis.core.message.impl.CoreMessageObjectPools;
+import org.apache.activemq.artemis.core.persistence.Persister;
+import org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl;
+import org.apache.activemq.artemis.reader.MessageUtil;
+import org.apache.activemq.artemis.utils.DataConstants;
+import org.apache.activemq.artemis.utils.UUID;
+import org.apache.activemq.artemis.utils.collections.TypedProperties;
+import org.jboss.logging.Logger;
+
+/** Note: you shouldn't change properties using multi-threads. Change your properties before you can send it to multiple
+ * consumers */
+public class OpenWireMessage extends RefCountMessage {
+ public static final int BUFFER_HEADER_SPACE = PacketImpl.PACKET_HEADERS_SIZE;
+
+ private volatile int memoryEstimate = -1;
+ private static final Logger logger = Logger.getLogger(OpenWireMessage.class);
+
+ // There's an integer with the number of bytes for the body
+ public static final int BODY_OFFSET = DataConstants.SIZE_INT;
+
+ /** That is the encode for the whole message, including properties..
+ it does not include the buffer for the Packet send and receive header on core protocol */
+ protected ByteBuf buffer;
+
+ private volatile boolean validBuffer = false;
--- End diff --
optional but (very) wellcome: `volatile` and `synchronized` has a big cost on hot paths (even if uncontended) so please evaluate which properties/methods could avoid using them on hot paths.
I know that the original current `CoreMessage` hasn't fixed yet on this aspect, but it would be great to have this one right from the start :+1: @michaelandrepearce wdyt?
---
[GitHub] activemq-artemis issue #1793: ARTEMIS-1498: Openwire internal headers should...
Posted by franz1981 <gi...@git.apache.org>.
Github user franz1981 commented on the issue:
https://github.com/apache/activemq-artemis/pull/1793
Please check this too https://github.com/apache/activemq-artemis/pull/1786
Probably will be easier to wait mine to be merged: the changes I've pushed are including some refactoring that will make this change to be easier, wdyt?
---
[GitHub] activemq-artemis pull request #1793: ARTEMIS-1498: Openwire internal headers...
Posted by franz1981 <gi...@git.apache.org>.
Github user franz1981 commented on a diff in the pull request:
https://github.com/apache/activemq-artemis/pull/1793#discussion_r165330595
--- Diff: artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireMessagePersister.java ---
@@ -0,0 +1,69 @@
+/*
+ * 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.activemq.artemis.core.protocol.openwire;
+
+import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
+import org.apache.activemq.artemis.api.core.Message;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.persistence.Persister;
+import org.apache.activemq.artemis.utils.DataConstants;
+
+public class OpenWireMessagePersister implements Persister<Message> {
--- End diff --
The same comments of `OpenWireConverter`
---
[GitHub] activemq-artemis pull request #1793: ARTEMIS-1498: Openwire internal headers...
Posted by michaelandrepearce <gi...@git.apache.org>.
Github user michaelandrepearce commented on a diff in the pull request:
https://github.com/apache/activemq-artemis/pull/1793#discussion_r165331496
--- Diff: artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireCoreConverter.java ---
@@ -0,0 +1,572 @@
+/*
+ * 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.activemq.artemis.core.protocol.openwire;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Arrays;
+import java.util.Map;
+import java.util.Set;
+import java.util.Map.Entry;
+import java.util.zip.Deflater;
+import java.util.zip.DeflaterOutputStream;
+import java.util.zip.Inflater;
+import java.util.zip.InflaterInputStream;
+import java.util.zip.InflaterOutputStream;
+
+import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
+import org.apache.activemq.artemis.api.core.FilterConstants;
+import org.apache.activemq.artemis.api.core.ICoreMessage;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.message.impl.CoreMessage;
+import org.apache.activemq.artemis.core.message.impl.CoreMessageObjectPools;
+import org.apache.activemq.artemis.reader.MessageUtil;
+import org.apache.activemq.artemis.utils.DataConstants;
+import org.apache.activemq.artemis.utils.collections.TypedProperties;
+import org.apache.activemq.command.CommandTypes;
+import org.apache.activemq.util.ByteArrayInputStream;
+import org.apache.activemq.util.ByteSequence;
+import org.apache.activemq.util.ByteSequenceData;
+import org.apache.activemq.util.MarshallingSupport;
+import org.fusesource.hawtbuf.UTF8Buffer;
+
+public class OpenWireCoreConverter {
+ public static final SimpleString JMS_GROUPSEQ = new SimpleString("JMSXGroupSequence");
+ public static final SimpleString JMS_XGROUPSEQ = new SimpleString("JMSXGroupSeq");
+ public static final SimpleString JMS_GROUPID = new SimpleString("JMSXGroupID");
+ private static final SimpleString AMQP_REPLYTOGROUPID = new SimpleString("JMS_AMQP_ReplyToGroupID");
+ public static final SimpleString JMS_CONTENT_TYPE = new SimpleString("JMS_AMQP_ContentType");
+ private static final SimpleString AMQP_CONTETTYPE = new SimpleString("JMS_AMQP_CONTENT_TYPE");
+
+ public static ICoreMessage toCore(final OpenWireMessage openwireMessage, final CoreMessageObjectPools coreMessageObjectPools)
+ throws Exception {
+ CoreMessage coreMessage = new CoreMessage(-1, openwireMessage.getMessageSize(), coreMessageObjectPools);
+ final SimpleString type = openwireMessage.getSimpleStringProperty(new SimpleString("JMSType"));
+ if (type != null) {
+ coreMessage.putStringProperty(SimpleString.toSimpleString("JMSType"), type);
--- End diff --
Or just make it a constant.
---
[GitHub] activemq-artemis pull request #1793: ARTEMIS-1498: Openwire internal headers...
Posted by michaelandrepearce <gi...@git.apache.org>.
Github user michaelandrepearce commented on a diff in the pull request:
https://github.com/apache/activemq-artemis/pull/1793#discussion_r168166219
--- Diff: artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireMessage.java ---
@@ -0,0 +1,673 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.activemq.artemis.core.protocol.openwire;
+
+import javax.jms.JMSException;
+import java.io.IOException;
+import java.util.Enumeration;
+import java.util.HashSet;
+import java.util.Set;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
+import org.apache.activemq.artemis.api.core.ActiveMQException;
+import org.apache.activemq.artemis.api.core.ActiveMQPropertyConversionException;
+import org.apache.activemq.artemis.api.core.ICoreMessage;
+import org.apache.activemq.artemis.api.core.Message;
+import org.apache.activemq.artemis.api.core.RefCountMessage;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.buffers.impl.ChannelBufferWrapper;
+import org.apache.activemq.artemis.core.message.impl.CoreMessageObjectPools;
+import org.apache.activemq.artemis.core.persistence.Persister;
+import org.apache.activemq.artemis.utils.DataConstants;
+import org.apache.activemq.command.ActiveMQDestination;
+import org.apache.activemq.command.ActiveMQMessage;
+import org.apache.activemq.openwire.OpenWireFormat;
+import org.apache.activemq.util.ByteSequence;
+import org.apache.activemq.wireformat.WireFormat;
+
+public class OpenWireMessage extends RefCountMessage {
+
+ private org.apache.activemq.command.ActiveMQMessage message;
+ private WireFormat wireFormat;
+ ByteBuf data;
+ boolean bufferValid;
+
+
+ public OpenWireMessage(org.apache.activemq.command.ActiveMQMessage message, WireFormat wireFormat) {
+ this.message = message;
+ this.wireFormat = wireFormat;
+ try {
+ ByteSequence byteSequence = this.wireFormat.marshal(message);
+ setBuffer(Unpooled.copiedBuffer(byteSequence.getData(), byteSequence.getOffset(), byteSequence.getLength()));
+ } catch (IOException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ public OpenWireMessage() {
+ }
+
+ public void setMarsheller(WireFormat wireFormat) {
+ this.wireFormat = wireFormat;
+ }
+
+ public ActiveMQMessage getAMQMessage() {
+ if (message == null) {
+ if (data != null) {
+ try {
+ message = (ActiveMQMessage) wireFormat.unmarshal(new ChannelBufferWrapper(data));
+ } catch (IOException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return message;
+ } else {
+ return new ActiveMQMessage();
+ }
+ }
+ return message;
+ }
+
+ @Override
+ public void messageChanged() {
+ }
+
+ @Override
+ public Long getScheduledDeliveryTime() {
+ return message.getArrival();
+ }
+
+ @Override
+ public SimpleString getReplyTo() {
+ return SimpleString.toSimpleString(message.getReplyTo().getPhysicalName());
+ }
+
+ @Override
+ public Message setReplyTo(SimpleString address) {
+ message.setReplyTo(ActiveMQDestination.createDestination(address.toString(), ActiveMQDestination.QUEUE_TYPE));
+ return this;
+ }
+
+ @Override
+ public Message setBuffer(ByteBuf buffer) {
+ this.data = buffer;
+ return this;
+ }
+
+ @Override
+ public ByteBuf getBuffer() {
+ if (data == null) {
+ return null;
+ } else {
+ return Unpooled.wrappedBuffer(data);
+ }
+ }
+
+ @Override
+ public OpenWireMessage copy() {
+ return new OpenWireMessage((ActiveMQMessage)message.copy(), wireFormat);
+ }
+
+ @Override
+ public Message copy(long newID) {
+ OpenWireMessage copy = new OpenWireMessage((ActiveMQMessage)message.copy(), wireFormat);
+ copy.setMessageID(newID);
+ return copy;
+ }
+
+ @Override
+ public long getMessageID() {
+ return message.getMessageId().getBrokerSequenceId();
+ }
+
+ @Override
+ public Message setMessageID(long id) {
+ message.getMessageId().setBrokerSequenceId(id);
+ return this;
+ }
+
+ @Override
+ public long getExpiration() {
+ return message.getExpiration();
+ }
+
+ @Override
+ public Message setExpiration(long expiration) {
+ message.setExpiration(expiration);
+ return this;
+ }
+
+ @Override
+ public Object getUserID() {
+ return message.getUserID();
+ }
+
+ @Override
+ public Message setUserID(Object userID) {
+ message.setUserID(userID.toString());
+ return this;
+ }
+
+ @Override
+ public boolean isDurable() {
+ return message.isPersistent();
+ }
+
+ @Override
+ public Message setDurable(boolean durable) {
+ message.setPersistent(durable);
+ return this;
+ }
+
+ @Override
+ public Persister<Message> getPersister() {
+ return OpenWireMessagePersister.INSTANCE;
+ }
+
+ @Override
+ public String getAddress() {
+ return message.getDestination().getPhysicalName();
+ }
+
+ @Override
+ public Message setAddress(String address) {
+ message.setDestination(ActiveMQDestination.createDestination(address, ActiveMQDestination.QUEUE_TYPE));
+ return this;
+ }
+
+ @Override
+ public SimpleString getAddressSimpleString() {
+ return SimpleString.toSimpleString(message.getDestination().getPhysicalName());
+ }
+
+ @Override
+ public Message setAddress(SimpleString address) {
+ message.setDestination(ActiveMQDestination.createDestination(address.toString(), ActiveMQDestination.QUEUE_TYPE));
+ return this;
+ }
+
+ @Override
+ public long getTimestamp() {
+ return message.getTimestamp();
+ }
+
+ @Override
+ public Message setTimestamp(long timestamp) {
+ message.setTimestamp(timestamp);
+ return this;
+ }
+
+ @Override
+ public byte getPriority() {
+ return message.getPriority();
+ }
+
+ @Override
+ public Message setPriority(byte priority) {
+ message.setPriority(priority);
+ return this;
+ }
+
+ @Override
+ public void receiveBuffer(ByteBuf buffer) {
+
+ }
+
+ @Override
+ public void sendBuffer(ByteBuf buffer, int deliveryCount) {
+
+ }
+
+ @Override
+ public int getPersistSize() {
+ return DataConstants.SIZE_INT + data.array().length;
+ }
+
+ @Override
+ public void persist(ActiveMQBuffer targetRecord) {
+ targetRecord.writeInt(data.writerIndex());
+ targetRecord.writeBytes(data, 0, data.writerIndex());
+ }
+
+ @Override
+ public void reloadPersistence(ActiveMQBuffer record) {
+ int size = record.readInt();
+ byte[] recordArray = new byte[size];
+ record.readBytes(recordArray);
+ this.data = Unpooled.wrappedBuffer(recordArray);
+ OpenWireFormat owf = new OpenWireFormat();
+ try {
+ message = (ActiveMQMessage) owf.unmarshal(new ChannelBufferWrapper(data));
+ } catch (IOException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Message putBooleanProperty(String key, boolean value) {
+ try {
+ message.setBooleanProperty(key, value);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putByteProperty(String key, byte value) {
+ try {
+ message.setByteProperty(key, value);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putBytesProperty(String key, byte[] value) {
+ return putObjectProperty(key, value);
+ }
+
+ @Override
+ public Message putShortProperty(String key, short value) {
+ try {
+ message.setShortProperty(key, value);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putCharProperty(String key, char value) {
+ return putStringProperty(key, Character.toString(value));
+ }
+
+ @Override
+ public Message putIntProperty(String key, int value) {
+ try {
+ message.setIntProperty(key, value);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putLongProperty(String key, long value) {
+ try {
+ message.setLongProperty(key, value);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putFloatProperty(String key, float value) {
+ try {
+ message.setFloatProperty(key, value);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putDoubleProperty(String key, double value) {
+ try {
+ message.setDoubleProperty(key, value);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putBooleanProperty(SimpleString key, boolean value) {
+ return putBooleanProperty(key.toString(), value);
+ }
+
+ @Override
+ public Message putByteProperty(SimpleString key, byte value) {
+ return putByteProperty(key.toString(), value);
+ }
+
+ @Override
+ public Message putBytesProperty(SimpleString key, byte[] value) {
+ return putBytesProperty(key.toString(), value);
+ }
+
+ @Override
+ public Message putShortProperty(SimpleString key, short value) {
+ return putShortProperty(key.toString(), value);
+ }
+
+ @Override
+ public Message putCharProperty(SimpleString key, char value) {
+ return putCharProperty(key.toString(), value);
+ }
+
+ @Override
+ public Message putIntProperty(SimpleString key, int value) {
+ return putIntProperty(key.toString(), value);
+ }
+
+ @Override
+ public Message putLongProperty(SimpleString key, long value) {
+ return putLongProperty(key.toString(), value);
+ }
+
+ @Override
+ public Message putFloatProperty(SimpleString key, float value) {
+ return putFloatProperty(key.toString(), value);
+
+ }
+
+ @Override
+ public Message putDoubleProperty(SimpleString key, double value) {
+ return putDoubleProperty(key.toString(), value);
+ }
+
+ @Override
+ public Message putStringProperty(String key, String value) {
+ try {
+ message.setStringProperty(key, value);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putObjectProperty(String key, Object value) throws ActiveMQPropertyConversionException {
+ try {
+ final Object v;
+ if (value instanceof SimpleString) {
+ v = value.toString();
+ } else {
+ v = value;
+ }
+ message.setProperty(key, value);
+ } catch (IOException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putObjectProperty(SimpleString key, Object value) throws ActiveMQPropertyConversionException {
+ return putObjectProperty(key.toString(), value);
+ }
+
+ @Override
+ public Object removeProperty(String key) {
+ try {
+ Object o = message.getProperty(key);
+ if (o != null) {
+ message.removeProperty(key);
+ }
+ return o;
+ } catch (IOException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+
+ }
+
+ @Override
+ public boolean containsProperty(String key) {
+ try {
+ return message.getProperty(key) != null;
+ } catch (IOException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Boolean getBooleanProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return message.getBooleanProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Byte getByteProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return message.getByteProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Double getDoubleProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return message.getDoubleProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Integer getIntProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return message.getIntProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Long getLongProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return message.getLongProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Object getObjectProperty(String key) {
+ try {
+ return message.getObjectProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Short getShortProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return message.getShortProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Float getFloatProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return message.getFloatProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public String getStringProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return message.getStringProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public SimpleString getSimpleStringProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return SimpleString.toSimpleString(message.getStringProperty(key));
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public byte[] getBytesProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return (byte[]) message.getObjectProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Object removeProperty(SimpleString key) {
+ return removeProperty(key.toString());
+ }
+
+ @Override
+ public boolean containsProperty(SimpleString key) {
+ return containsProperty(key.toString());
+ }
+
+ @Override
+ public Boolean getBooleanProperty(SimpleString key) throws ActiveMQPropertyConversionException {
+ return getBooleanProperty(key.toString());
+ }
+
+ @Override
+ public Byte getByteProperty(SimpleString key) throws ActiveMQPropertyConversionException {
+ return getByteProperty(key.toString());
+ }
+
+ @Override
+ public Double getDoubleProperty(SimpleString key) throws ActiveMQPropertyConversionException {
+ return getDoubleProperty(key.toString());
+ }
+
+ @Override
+ public Integer getIntProperty(SimpleString key) throws ActiveMQPropertyConversionException {
+ return getIntProperty(key.toString());
+ }
+
+ @Override
+ public Long getLongProperty(SimpleString key) throws ActiveMQPropertyConversionException {
+ return getLongProperty(key.toString());
+ }
+
+ @Override
+ public Object getObjectProperty(SimpleString key) {
+ return getObjectProperty(key.toString());
+ }
+
+ @Override
+ public Object getAnnotation(SimpleString key) {
+ return getObjectProperty(key.toString());
+ }
+
+ @Override
+ public Short getShortProperty(SimpleString key) throws ActiveMQPropertyConversionException {
+ return getShortProperty(key.toString());
+ }
+
+ @Override
+ public Float getFloatProperty(SimpleString key) throws ActiveMQPropertyConversionException {
+ return getFloatProperty(key.toString());
+ }
+
+ @Override
+ public String getStringProperty(SimpleString key) throws ActiveMQPropertyConversionException {
+ return getStringProperty(key.toString());
+ }
+
+ @Override
+ public SimpleString getSimpleStringProperty(SimpleString key) throws ActiveMQPropertyConversionException {
+ return SimpleString.toSimpleString(getStringProperty(key));
+ }
+
+ @Override
+ public byte[] getBytesProperty(SimpleString key) throws ActiveMQPropertyConversionException {
+ return getBytesProperty(key.toString());
+ }
+
+ @Override
+ public Message putStringProperty(SimpleString key, SimpleString value) {
+ return putStringProperty(key.toString(), value.toString());
+ }
+
+ @Override
+ public Message putStringProperty(SimpleString key, String value) {
+ return putStringProperty(key.toString(), value);
+ }
+
+ @Override
+ public int getEncodeSize() {
+ return 0;
--- End diff --
Should be given/calculated
---
[GitHub] activemq-artemis pull request #1793: ARTEMIS-1498: Openwire internal headers...
Posted by michaelandrepearce <gi...@git.apache.org>.
Github user michaelandrepearce commented on a diff in the pull request:
https://github.com/apache/activemq-artemis/pull/1793#discussion_r168166856
--- Diff: artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireMessage.java ---
@@ -0,0 +1,673 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.activemq.artemis.core.protocol.openwire;
+
+import javax.jms.JMSException;
+import java.io.IOException;
+import java.util.Enumeration;
+import java.util.HashSet;
+import java.util.Set;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
+import org.apache.activemq.artemis.api.core.ActiveMQException;
+import org.apache.activemq.artemis.api.core.ActiveMQPropertyConversionException;
+import org.apache.activemq.artemis.api.core.ICoreMessage;
+import org.apache.activemq.artemis.api.core.Message;
+import org.apache.activemq.artemis.api.core.RefCountMessage;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.buffers.impl.ChannelBufferWrapper;
+import org.apache.activemq.artemis.core.message.impl.CoreMessageObjectPools;
+import org.apache.activemq.artemis.core.persistence.Persister;
+import org.apache.activemq.artemis.utils.DataConstants;
+import org.apache.activemq.command.ActiveMQDestination;
+import org.apache.activemq.command.ActiveMQMessage;
+import org.apache.activemq.openwire.OpenWireFormat;
+import org.apache.activemq.util.ByteSequence;
+import org.apache.activemq.wireformat.WireFormat;
+
+public class OpenWireMessage extends RefCountMessage {
+
+ private org.apache.activemq.command.ActiveMQMessage message;
+ private WireFormat wireFormat;
+ ByteBuf data;
+ boolean bufferValid;
+
+
+ public OpenWireMessage(org.apache.activemq.command.ActiveMQMessage message, WireFormat wireFormat) {
+ this.message = message;
+ this.wireFormat = wireFormat;
+ try {
+ ByteSequence byteSequence = this.wireFormat.marshal(message);
+ setBuffer(Unpooled.copiedBuffer(byteSequence.getData(), byteSequence.getOffset(), byteSequence.getLength()));
+ } catch (IOException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ public OpenWireMessage() {
+ }
+
+ public void setMarsheller(WireFormat wireFormat) {
+ this.wireFormat = wireFormat;
+ }
+
+ public ActiveMQMessage getAMQMessage() {
+ if (message == null) {
+ if (data != null) {
+ try {
+ message = (ActiveMQMessage) wireFormat.unmarshal(new ChannelBufferWrapper(data));
+ } catch (IOException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return message;
+ } else {
+ return new ActiveMQMessage();
+ }
+ }
+ return message;
+ }
+
+ @Override
+ public void messageChanged() {
+ }
+
+ @Override
+ public Long getScheduledDeliveryTime() {
+ return message.getArrival();
+ }
+
+ @Override
+ public SimpleString getReplyTo() {
+ return SimpleString.toSimpleString(message.getReplyTo().getPhysicalName());
+ }
+
+ @Override
+ public Message setReplyTo(SimpleString address) {
+ message.setReplyTo(ActiveMQDestination.createDestination(address.toString(), ActiveMQDestination.QUEUE_TYPE));
+ return this;
+ }
+
+ @Override
+ public Message setBuffer(ByteBuf buffer) {
+ this.data = buffer;
+ return this;
+ }
+
+ @Override
+ public ByteBuf getBuffer() {
+ if (data == null) {
+ return null;
+ } else {
+ return Unpooled.wrappedBuffer(data);
+ }
+ }
+
+ @Override
+ public OpenWireMessage copy() {
+ return new OpenWireMessage((ActiveMQMessage)message.copy(), wireFormat);
+ }
+
+ @Override
+ public Message copy(long newID) {
+ OpenWireMessage copy = new OpenWireMessage((ActiveMQMessage)message.copy(), wireFormat);
+ copy.setMessageID(newID);
+ return copy;
+ }
+
+ @Override
+ public long getMessageID() {
+ return message.getMessageId().getBrokerSequenceId();
+ }
+
+ @Override
+ public Message setMessageID(long id) {
+ message.getMessageId().setBrokerSequenceId(id);
+ return this;
+ }
+
+ @Override
+ public long getExpiration() {
+ return message.getExpiration();
+ }
+
+ @Override
+ public Message setExpiration(long expiration) {
+ message.setExpiration(expiration);
+ return this;
+ }
+
+ @Override
+ public Object getUserID() {
+ return message.getUserID();
+ }
+
+ @Override
+ public Message setUserID(Object userID) {
+ message.setUserID(userID.toString());
+ return this;
+ }
+
+ @Override
+ public boolean isDurable() {
+ return message.isPersistent();
+ }
+
+ @Override
+ public Message setDurable(boolean durable) {
+ message.setPersistent(durable);
+ return this;
+ }
+
+ @Override
+ public Persister<Message> getPersister() {
+ return OpenWireMessagePersister.INSTANCE;
+ }
+
+ @Override
+ public String getAddress() {
+ return message.getDestination().getPhysicalName();
+ }
+
+ @Override
+ public Message setAddress(String address) {
+ message.setDestination(ActiveMQDestination.createDestination(address, ActiveMQDestination.QUEUE_TYPE));
+ return this;
+ }
+
+ @Override
+ public SimpleString getAddressSimpleString() {
+ return SimpleString.toSimpleString(message.getDestination().getPhysicalName());
+ }
+
+ @Override
+ public Message setAddress(SimpleString address) {
+ message.setDestination(ActiveMQDestination.createDestination(address.toString(), ActiveMQDestination.QUEUE_TYPE));
+ return this;
+ }
+
+ @Override
+ public long getTimestamp() {
+ return message.getTimestamp();
+ }
+
+ @Override
+ public Message setTimestamp(long timestamp) {
+ message.setTimestamp(timestamp);
+ return this;
+ }
+
+ @Override
+ public byte getPriority() {
+ return message.getPriority();
+ }
+
+ @Override
+ public Message setPriority(byte priority) {
+ message.setPriority(priority);
+ return this;
+ }
+
+ @Override
+ public void receiveBuffer(ByteBuf buffer) {
+
+ }
+
+ @Override
+ public void sendBuffer(ByteBuf buffer, int deliveryCount) {
+
+ }
+
+ @Override
+ public int getPersistSize() {
+ return DataConstants.SIZE_INT + data.array().length;
+ }
+
+ @Override
+ public void persist(ActiveMQBuffer targetRecord) {
+ targetRecord.writeInt(data.writerIndex());
+ targetRecord.writeBytes(data, 0, data.writerIndex());
+ }
+
+ @Override
+ public void reloadPersistence(ActiveMQBuffer record) {
+ int size = record.readInt();
+ byte[] recordArray = new byte[size];
+ record.readBytes(recordArray);
+ this.data = Unpooled.wrappedBuffer(recordArray);
+ OpenWireFormat owf = new OpenWireFormat();
+ try {
+ message = (ActiveMQMessage) owf.unmarshal(new ChannelBufferWrapper(data));
+ } catch (IOException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Message putBooleanProperty(String key, boolean value) {
+ try {
+ message.setBooleanProperty(key, value);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putByteProperty(String key, byte value) {
+ try {
+ message.setByteProperty(key, value);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putBytesProperty(String key, byte[] value) {
+ return putObjectProperty(key, value);
+ }
+
+ @Override
+ public Message putShortProperty(String key, short value) {
+ try {
+ message.setShortProperty(key, value);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putCharProperty(String key, char value) {
+ return putStringProperty(key, Character.toString(value));
+ }
+
+ @Override
+ public Message putIntProperty(String key, int value) {
+ try {
+ message.setIntProperty(key, value);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putLongProperty(String key, long value) {
+ try {
+ message.setLongProperty(key, value);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putFloatProperty(String key, float value) {
+ try {
+ message.setFloatProperty(key, value);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putDoubleProperty(String key, double value) {
+ try {
+ message.setDoubleProperty(key, value);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putBooleanProperty(SimpleString key, boolean value) {
+ return putBooleanProperty(key.toString(), value);
+ }
+
+ @Override
+ public Message putByteProperty(SimpleString key, byte value) {
+ return putByteProperty(key.toString(), value);
+ }
+
+ @Override
+ public Message putBytesProperty(SimpleString key, byte[] value) {
+ return putBytesProperty(key.toString(), value);
+ }
+
+ @Override
+ public Message putShortProperty(SimpleString key, short value) {
+ return putShortProperty(key.toString(), value);
+ }
+
+ @Override
+ public Message putCharProperty(SimpleString key, char value) {
+ return putCharProperty(key.toString(), value);
+ }
+
+ @Override
+ public Message putIntProperty(SimpleString key, int value) {
+ return putIntProperty(key.toString(), value);
+ }
+
+ @Override
+ public Message putLongProperty(SimpleString key, long value) {
+ return putLongProperty(key.toString(), value);
+ }
+
+ @Override
+ public Message putFloatProperty(SimpleString key, float value) {
+ return putFloatProperty(key.toString(), value);
+
+ }
+
+ @Override
+ public Message putDoubleProperty(SimpleString key, double value) {
+ return putDoubleProperty(key.toString(), value);
+ }
+
+ @Override
+ public Message putStringProperty(String key, String value) {
+ try {
+ message.setStringProperty(key, value);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putObjectProperty(String key, Object value) throws ActiveMQPropertyConversionException {
+ try {
+ final Object v;
+ if (value instanceof SimpleString) {
+ v = value.toString();
+ } else {
+ v = value;
+ }
+ message.setProperty(key, value);
+ } catch (IOException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putObjectProperty(SimpleString key, Object value) throws ActiveMQPropertyConversionException {
+ return putObjectProperty(key.toString(), value);
+ }
+
+ @Override
+ public Object removeProperty(String key) {
+ try {
+ Object o = message.getProperty(key);
+ if (o != null) {
+ message.removeProperty(key);
+ }
+ return o;
+ } catch (IOException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+
+ }
+
+ @Override
+ public boolean containsProperty(String key) {
+ try {
+ return message.getProperty(key) != null;
+ } catch (IOException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Boolean getBooleanProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return message.getBooleanProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Byte getByteProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return message.getByteProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Double getDoubleProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return message.getDoubleProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Integer getIntProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return message.getIntProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Long getLongProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return message.getLongProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Object getObjectProperty(String key) {
+ try {
+ return message.getObjectProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Short getShortProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return message.getShortProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Float getFloatProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return message.getFloatProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public String getStringProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return message.getStringProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public SimpleString getSimpleStringProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return SimpleString.toSimpleString(message.getStringProperty(key));
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public byte[] getBytesProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return (byte[]) message.getObjectProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Object removeProperty(SimpleString key) {
+ return removeProperty(key.toString());
+ }
+
+ @Override
+ public boolean containsProperty(SimpleString key) {
+ return containsProperty(key.toString());
+ }
+
+ @Override
+ public Boolean getBooleanProperty(SimpleString key) throws ActiveMQPropertyConversionException {
+ return getBooleanProperty(key.toString());
+ }
+
+ @Override
+ public Byte getByteProperty(SimpleString key) throws ActiveMQPropertyConversionException {
+ return getByteProperty(key.toString());
+ }
+
+ @Override
+ public Double getDoubleProperty(SimpleString key) throws ActiveMQPropertyConversionException {
+ return getDoubleProperty(key.toString());
+ }
+
+ @Override
+ public Integer getIntProperty(SimpleString key) throws ActiveMQPropertyConversionException {
+ return getIntProperty(key.toString());
+ }
+
+ @Override
+ public Long getLongProperty(SimpleString key) throws ActiveMQPropertyConversionException {
+ return getLongProperty(key.toString());
+ }
+
+ @Override
+ public Object getObjectProperty(SimpleString key) {
+ return getObjectProperty(key.toString());
+ }
+
+ @Override
+ public Object getAnnotation(SimpleString key) {
+ return getObjectProperty(key.toString());
+ }
+
+ @Override
+ public Short getShortProperty(SimpleString key) throws ActiveMQPropertyConversionException {
+ return getShortProperty(key.toString());
+ }
+
+ @Override
+ public Float getFloatProperty(SimpleString key) throws ActiveMQPropertyConversionException {
+ return getFloatProperty(key.toString());
+ }
+
+ @Override
+ public String getStringProperty(SimpleString key) throws ActiveMQPropertyConversionException {
+ return getStringProperty(key.toString());
+ }
+
+ @Override
+ public SimpleString getSimpleStringProperty(SimpleString key) throws ActiveMQPropertyConversionException {
+ return SimpleString.toSimpleString(getStringProperty(key));
+ }
+
+ @Override
+ public byte[] getBytesProperty(SimpleString key) throws ActiveMQPropertyConversionException {
+ return getBytesProperty(key.toString());
+ }
+
+ @Override
+ public Message putStringProperty(SimpleString key, SimpleString value) {
+ return putStringProperty(key.toString(), value.toString());
+ }
+
+ @Override
+ public Message putStringProperty(SimpleString key, String value) {
+ return putStringProperty(key.toString(), value);
+ }
+
+ @Override
+ public int getEncodeSize() {
+ return 0;
+ }
+
+ @Override
+ public Set<SimpleString> getPropertyNames() {
+ try {
+ Set<SimpleString> simpleStrings = new HashSet<>();
+ Enumeration propertyNames = message.getPropertyNames();
+ while (propertyNames.hasMoreElements()) {
+ simpleStrings.add(SimpleString.toSimpleString((String) propertyNames.nextElement()));
+ }
+ return simpleStrings;
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public ICoreMessage toCore() {
+ return toCore(null);
+ }
+
+ @Override
+ public ICoreMessage toCore(CoreMessageObjectPools coreMessageObjectPools) {
+ try {
+ return OpenWireMessageConverter.convertAMQToCore(message, coreMessageObjectPools);
+ } catch (Exception e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public int getMemoryEstimate() {
+ return message.getSize();
+ }
+
+ @Override
+ public Object getDuplicateProperty() {
+ return getObjectProperty(Message.HDR_DUPLICATE_DETECTION_ID);
+ }
+
+ @Override
+ public SimpleString getLastValueProperty() {
+ return getSimpleStringProperty(Message.HDR_LAST_VALUE_NAME);
--- End diff --
Use pool
---
[GitHub] activemq-artemis issue #1793: ARTEMIS-1498: Openwire internal headers should...
Posted by franz1981 <gi...@git.apache.org>.
Github user franz1981 commented on the issue:
https://github.com/apache/activemq-artemis/pull/1793
@clebertsuconic @RaiSaurabh @michaelandrepearce
**Re performances**
I've the first results using just a small benchmark with 1 producer/1 consumer using 100 bytes messages.
```
this PR: 37050 msg/sec
2.5: 85689 msg/sec
```
So performance-wise there is a long way to go, but I think it is related to some correctness issues.
**Re correctness**
I've launched a run on CI just with the OpenWire test suite with no success:
```
[INFO] ActiveMQ5.x unit tests ............................. FAILURE [ 2.950 s]
```
So I've decided to run the most some on my box@home.
The results from `org.apache.activemq.artemis.tests.integration.openwire` ones shows 7 failed tests/341:
```
CompressedInteropTest.testCoreReceiveOpenWireCompressedMessages
MessageListenerRedeliveryTest.testQueueSessionListenerExceptionDlq
OpenWireLargeMessageTest.testSendReceiveLargeMessage
GeneralInteropTest.testSendingToCoreJms
SimpleOpenWireTest.testTempQueueSendAfterConnectionClose
RedeliveryPolicyTest.testRedeliveryPolicyPerDestination
RedeliveryPolicyTest.testDLQHandling
```
I've to run the `tests/activemq5-unit-tests` yet but I suppose that's enough to be fixed.
Sorry for letting this PR sitting here for such a long time and thanks @RaiSaurabh to have worked on it: I think that given the right direction taken by the work on this PR makes a lot of sense to fix all the compatibility issues first, before looking to the performance gains that in the current state, are not present. @RaiSaurabh don't give up with the hard work done here and take a look at the tests results: as much as possible I will try to help by reviewing the code and propose fixes :+1:
---
[GitHub] activemq-artemis pull request #1793: ARTEMIS-1498: Openwire internal headers...
Posted by michaelandrepearce <gi...@git.apache.org>.
Github user michaelandrepearce commented on a diff in the pull request:
https://github.com/apache/activemq-artemis/pull/1793#discussion_r168166679
--- Diff: artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireMessage.java ---
@@ -0,0 +1,673 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.activemq.artemis.core.protocol.openwire;
+
+import javax.jms.JMSException;
+import java.io.IOException;
+import java.util.Enumeration;
+import java.util.HashSet;
+import java.util.Set;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
+import org.apache.activemq.artemis.api.core.ActiveMQException;
+import org.apache.activemq.artemis.api.core.ActiveMQPropertyConversionException;
+import org.apache.activemq.artemis.api.core.ICoreMessage;
+import org.apache.activemq.artemis.api.core.Message;
+import org.apache.activemq.artemis.api.core.RefCountMessage;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.buffers.impl.ChannelBufferWrapper;
+import org.apache.activemq.artemis.core.message.impl.CoreMessageObjectPools;
+import org.apache.activemq.artemis.core.persistence.Persister;
+import org.apache.activemq.artemis.utils.DataConstants;
+import org.apache.activemq.command.ActiveMQDestination;
+import org.apache.activemq.command.ActiveMQMessage;
+import org.apache.activemq.openwire.OpenWireFormat;
+import org.apache.activemq.util.ByteSequence;
+import org.apache.activemq.wireformat.WireFormat;
+
+public class OpenWireMessage extends RefCountMessage {
+
+ private org.apache.activemq.command.ActiveMQMessage message;
+ private WireFormat wireFormat;
+ ByteBuf data;
+ boolean bufferValid;
+
+
+ public OpenWireMessage(org.apache.activemq.command.ActiveMQMessage message, WireFormat wireFormat) {
+ this.message = message;
+ this.wireFormat = wireFormat;
+ try {
+ ByteSequence byteSequence = this.wireFormat.marshal(message);
+ setBuffer(Unpooled.copiedBuffer(byteSequence.getData(), byteSequence.getOffset(), byteSequence.getLength()));
+ } catch (IOException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ public OpenWireMessage() {
+ }
+
+ public void setMarsheller(WireFormat wireFormat) {
+ this.wireFormat = wireFormat;
+ }
+
+ public ActiveMQMessage getAMQMessage() {
+ if (message == null) {
+ if (data != null) {
+ try {
+ message = (ActiveMQMessage) wireFormat.unmarshal(new ChannelBufferWrapper(data));
+ } catch (IOException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return message;
+ } else {
+ return new ActiveMQMessage();
+ }
+ }
+ return message;
+ }
+
+ @Override
+ public void messageChanged() {
+ }
+
+ @Override
+ public Long getScheduledDeliveryTime() {
+ return message.getArrival();
+ }
+
+ @Override
+ public SimpleString getReplyTo() {
+ return SimpleString.toSimpleString(message.getReplyTo().getPhysicalName());
+ }
+
+ @Override
+ public Message setReplyTo(SimpleString address) {
+ message.setReplyTo(ActiveMQDestination.createDestination(address.toString(), ActiveMQDestination.QUEUE_TYPE));
+ return this;
+ }
+
+ @Override
+ public Message setBuffer(ByteBuf buffer) {
+ this.data = buffer;
+ return this;
+ }
+
+ @Override
+ public ByteBuf getBuffer() {
+ if (data == null) {
+ return null;
+ } else {
+ return Unpooled.wrappedBuffer(data);
+ }
+ }
+
+ @Override
+ public OpenWireMessage copy() {
+ return new OpenWireMessage((ActiveMQMessage)message.copy(), wireFormat);
+ }
+
+ @Override
+ public Message copy(long newID) {
+ OpenWireMessage copy = new OpenWireMessage((ActiveMQMessage)message.copy(), wireFormat);
+ copy.setMessageID(newID);
+ return copy;
+ }
+
+ @Override
+ public long getMessageID() {
+ return message.getMessageId().getBrokerSequenceId();
+ }
+
+ @Override
+ public Message setMessageID(long id) {
+ message.getMessageId().setBrokerSequenceId(id);
+ return this;
+ }
+
+ @Override
+ public long getExpiration() {
+ return message.getExpiration();
+ }
+
+ @Override
+ public Message setExpiration(long expiration) {
+ message.setExpiration(expiration);
+ return this;
+ }
+
+ @Override
+ public Object getUserID() {
+ return message.getUserID();
+ }
+
+ @Override
+ public Message setUserID(Object userID) {
+ message.setUserID(userID.toString());
+ return this;
+ }
+
+ @Override
+ public boolean isDurable() {
+ return message.isPersistent();
+ }
+
+ @Override
+ public Message setDurable(boolean durable) {
+ message.setPersistent(durable);
+ return this;
+ }
+
+ @Override
+ public Persister<Message> getPersister() {
+ return OpenWireMessagePersister.INSTANCE;
+ }
+
+ @Override
+ public String getAddress() {
+ return message.getDestination().getPhysicalName();
+ }
+
+ @Override
+ public Message setAddress(String address) {
+ message.setDestination(ActiveMQDestination.createDestination(address, ActiveMQDestination.QUEUE_TYPE));
+ return this;
+ }
+
+ @Override
+ public SimpleString getAddressSimpleString() {
+ return SimpleString.toSimpleString(message.getDestination().getPhysicalName());
+ }
+
+ @Override
+ public Message setAddress(SimpleString address) {
+ message.setDestination(ActiveMQDestination.createDestination(address.toString(), ActiveMQDestination.QUEUE_TYPE));
+ return this;
+ }
+
+ @Override
+ public long getTimestamp() {
+ return message.getTimestamp();
+ }
+
+ @Override
+ public Message setTimestamp(long timestamp) {
+ message.setTimestamp(timestamp);
+ return this;
+ }
+
+ @Override
+ public byte getPriority() {
+ return message.getPriority();
+ }
+
+ @Override
+ public Message setPriority(byte priority) {
+ message.setPriority(priority);
+ return this;
+ }
+
+ @Override
+ public void receiveBuffer(ByteBuf buffer) {
+
+ }
+
+ @Override
+ public void sendBuffer(ByteBuf buffer, int deliveryCount) {
+
+ }
+
+ @Override
+ public int getPersistSize() {
+ return DataConstants.SIZE_INT + data.array().length;
+ }
+
+ @Override
+ public void persist(ActiveMQBuffer targetRecord) {
+ targetRecord.writeInt(data.writerIndex());
+ targetRecord.writeBytes(data, 0, data.writerIndex());
+ }
+
+ @Override
+ public void reloadPersistence(ActiveMQBuffer record) {
+ int size = record.readInt();
+ byte[] recordArray = new byte[size];
+ record.readBytes(recordArray);
+ this.data = Unpooled.wrappedBuffer(recordArray);
+ OpenWireFormat owf = new OpenWireFormat();
+ try {
+ message = (ActiveMQMessage) owf.unmarshal(new ChannelBufferWrapper(data));
+ } catch (IOException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Message putBooleanProperty(String key, boolean value) {
+ try {
+ message.setBooleanProperty(key, value);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putByteProperty(String key, byte value) {
+ try {
+ message.setByteProperty(key, value);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putBytesProperty(String key, byte[] value) {
+ return putObjectProperty(key, value);
+ }
+
+ @Override
+ public Message putShortProperty(String key, short value) {
+ try {
+ message.setShortProperty(key, value);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putCharProperty(String key, char value) {
+ return putStringProperty(key, Character.toString(value));
+ }
+
+ @Override
+ public Message putIntProperty(String key, int value) {
+ try {
+ message.setIntProperty(key, value);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putLongProperty(String key, long value) {
+ try {
+ message.setLongProperty(key, value);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putFloatProperty(String key, float value) {
+ try {
+ message.setFloatProperty(key, value);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putDoubleProperty(String key, double value) {
+ try {
+ message.setDoubleProperty(key, value);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putBooleanProperty(SimpleString key, boolean value) {
+ return putBooleanProperty(key.toString(), value);
+ }
+
+ @Override
+ public Message putByteProperty(SimpleString key, byte value) {
+ return putByteProperty(key.toString(), value);
+ }
+
+ @Override
+ public Message putBytesProperty(SimpleString key, byte[] value) {
+ return putBytesProperty(key.toString(), value);
+ }
+
+ @Override
+ public Message putShortProperty(SimpleString key, short value) {
+ return putShortProperty(key.toString(), value);
+ }
+
+ @Override
+ public Message putCharProperty(SimpleString key, char value) {
+ return putCharProperty(key.toString(), value);
+ }
+
+ @Override
+ public Message putIntProperty(SimpleString key, int value) {
+ return putIntProperty(key.toString(), value);
+ }
+
+ @Override
+ public Message putLongProperty(SimpleString key, long value) {
+ return putLongProperty(key.toString(), value);
+ }
+
+ @Override
+ public Message putFloatProperty(SimpleString key, float value) {
+ return putFloatProperty(key.toString(), value);
+
+ }
+
+ @Override
+ public Message putDoubleProperty(SimpleString key, double value) {
+ return putDoubleProperty(key.toString(), value);
+ }
+
+ @Override
+ public Message putStringProperty(String key, String value) {
+ try {
+ message.setStringProperty(key, value);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putObjectProperty(String key, Object value) throws ActiveMQPropertyConversionException {
+ try {
+ final Object v;
+ if (value instanceof SimpleString) {
+ v = value.toString();
+ } else {
+ v = value;
+ }
+ message.setProperty(key, value);
+ } catch (IOException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putObjectProperty(SimpleString key, Object value) throws ActiveMQPropertyConversionException {
+ return putObjectProperty(key.toString(), value);
+ }
+
+ @Override
+ public Object removeProperty(String key) {
+ try {
+ Object o = message.getProperty(key);
+ if (o != null) {
+ message.removeProperty(key);
+ }
+ return o;
+ } catch (IOException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+
+ }
+
+ @Override
+ public boolean containsProperty(String key) {
+ try {
+ return message.getProperty(key) != null;
+ } catch (IOException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Boolean getBooleanProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return message.getBooleanProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Byte getByteProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return message.getByteProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Double getDoubleProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return message.getDoubleProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Integer getIntProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return message.getIntProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Long getLongProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return message.getLongProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Object getObjectProperty(String key) {
+ try {
+ return message.getObjectProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Short getShortProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return message.getShortProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Float getFloatProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return message.getFloatProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public String getStringProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return message.getStringProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public SimpleString getSimpleStringProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return SimpleString.toSimpleString(message.getStringProperty(key));
--- End diff --
Could use pool
---
[GitHub] activemq-artemis issue #1793: ARTEMIS-1498: Openwire internal headers should...
Posted by michaelandrepearce <gi...@git.apache.org>.
Github user michaelandrepearce commented on the issue:
https://github.com/apache/activemq-artemis/pull/1793
@RaiSaurabh @franz1981 I just got some perf results back on this, and its not pretty. (i retract my prev comments (i deleted)
After looking again it seems like all this is, is a clone of the CoreMessage, and still on send / inbound, its being converted from 5.x Message to this, instead of taking the underlying, as such its adding more overhead.
The intent of a native Message type, is that it holds the protocol specific objects internally as much as possible an operates/delegates to them, meaning that if on consume there was no changes or anything the same protocol specific object can be fired back out.
---
[GitHub] activemq-artemis pull request #1793: ARTEMIS-1498: Openwire internal headers...
Posted by michaelandrepearce <gi...@git.apache.org>.
Github user michaelandrepearce commented on a diff in the pull request:
https://github.com/apache/activemq-artemis/pull/1793#discussion_r168166047
--- Diff: artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireMessage.java ---
@@ -0,0 +1,673 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.activemq.artemis.core.protocol.openwire;
+
+import javax.jms.JMSException;
+import java.io.IOException;
+import java.util.Enumeration;
+import java.util.HashSet;
+import java.util.Set;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
+import org.apache.activemq.artemis.api.core.ActiveMQException;
+import org.apache.activemq.artemis.api.core.ActiveMQPropertyConversionException;
+import org.apache.activemq.artemis.api.core.ICoreMessage;
+import org.apache.activemq.artemis.api.core.Message;
+import org.apache.activemq.artemis.api.core.RefCountMessage;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.buffers.impl.ChannelBufferWrapper;
+import org.apache.activemq.artemis.core.message.impl.CoreMessageObjectPools;
+import org.apache.activemq.artemis.core.persistence.Persister;
+import org.apache.activemq.artemis.utils.DataConstants;
+import org.apache.activemq.command.ActiveMQDestination;
+import org.apache.activemq.command.ActiveMQMessage;
+import org.apache.activemq.openwire.OpenWireFormat;
+import org.apache.activemq.util.ByteSequence;
+import org.apache.activemq.wireformat.WireFormat;
+
+public class OpenWireMessage extends RefCountMessage {
+
+ private org.apache.activemq.command.ActiveMQMessage message;
+ private WireFormat wireFormat;
+ ByteBuf data;
+ boolean bufferValid;
+
+
+ public OpenWireMessage(org.apache.activemq.command.ActiveMQMessage message, WireFormat wireFormat) {
+ this.message = message;
+ this.wireFormat = wireFormat;
+ try {
+ ByteSequence byteSequence = this.wireFormat.marshal(message);
+ setBuffer(Unpooled.copiedBuffer(byteSequence.getData(), byteSequence.getOffset(), byteSequence.getLength()));
+ } catch (IOException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ public OpenWireMessage() {
+ }
+
+ public void setMarsheller(WireFormat wireFormat) {
+ this.wireFormat = wireFormat;
+ }
+
+ public ActiveMQMessage getAMQMessage() {
+ if (message == null) {
+ if (data != null) {
+ try {
+ message = (ActiveMQMessage) wireFormat.unmarshal(new ChannelBufferWrapper(data));
+ } catch (IOException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return message;
+ } else {
+ return new ActiveMQMessage();
+ }
+ }
+ return message;
+ }
+
+ @Override
+ public void messageChanged() {
+ }
+
+ @Override
+ public Long getScheduledDeliveryTime() {
+ return message.getArrival();
+ }
+
+ @Override
+ public SimpleString getReplyTo() {
+ return SimpleString.toSimpleString(message.getReplyTo().getPhysicalName());
+ }
+
+ @Override
+ public Message setReplyTo(SimpleString address) {
+ message.setReplyTo(ActiveMQDestination.createDestination(address.toString(), ActiveMQDestination.QUEUE_TYPE));
+ return this;
+ }
+
+ @Override
+ public Message setBuffer(ByteBuf buffer) {
+ this.data = buffer;
+ return this;
+ }
+
+ @Override
+ public ByteBuf getBuffer() {
+ if (data == null) {
+ return null;
+ } else {
+ return Unpooled.wrappedBuffer(data);
+ }
+ }
+
+ @Override
+ public OpenWireMessage copy() {
+ return new OpenWireMessage((ActiveMQMessage)message.copy(), wireFormat);
+ }
+
+ @Override
+ public Message copy(long newID) {
+ OpenWireMessage copy = new OpenWireMessage((ActiveMQMessage)message.copy(), wireFormat);
+ copy.setMessageID(newID);
+ return copy;
+ }
+
+ @Override
+ public long getMessageID() {
+ return message.getMessageId().getBrokerSequenceId();
+ }
+
+ @Override
+ public Message setMessageID(long id) {
+ message.getMessageId().setBrokerSequenceId(id);
+ return this;
+ }
+
+ @Override
+ public long getExpiration() {
+ return message.getExpiration();
+ }
+
+ @Override
+ public Message setExpiration(long expiration) {
+ message.setExpiration(expiration);
+ return this;
+ }
+
+ @Override
+ public Object getUserID() {
+ return message.getUserID();
+ }
+
+ @Override
+ public Message setUserID(Object userID) {
+ message.setUserID(userID.toString());
+ return this;
+ }
+
+ @Override
+ public boolean isDurable() {
+ return message.isPersistent();
+ }
+
+ @Override
+ public Message setDurable(boolean durable) {
+ message.setPersistent(durable);
+ return this;
+ }
+
+ @Override
+ public Persister<Message> getPersister() {
+ return OpenWireMessagePersister.INSTANCE;
+ }
+
+ @Override
+ public String getAddress() {
+ return message.getDestination().getPhysicalName();
+ }
+
+ @Override
+ public Message setAddress(String address) {
+ message.setDestination(ActiveMQDestination.createDestination(address, ActiveMQDestination.QUEUE_TYPE));
+ return this;
+ }
+
+ @Override
+ public SimpleString getAddressSimpleString() {
+ return SimpleString.toSimpleString(message.getDestination().getPhysicalName());
+ }
+
+ @Override
+ public Message setAddress(SimpleString address) {
+ message.setDestination(ActiveMQDestination.createDestination(address.toString(), ActiveMQDestination.QUEUE_TYPE));
+ return this;
+ }
+
+ @Override
+ public long getTimestamp() {
+ return message.getTimestamp();
+ }
+
+ @Override
+ public Message setTimestamp(long timestamp) {
+ message.setTimestamp(timestamp);
+ return this;
+ }
+
+ @Override
+ public byte getPriority() {
+ return message.getPriority();
+ }
+
+ @Override
+ public Message setPriority(byte priority) {
+ message.setPriority(priority);
+ return this;
+ }
+
+ @Override
+ public void receiveBuffer(ByteBuf buffer) {
+
--- End diff --
This should be implemented
---
[GitHub] activemq-artemis issue #1793: ARTEMIS-1498: Openwire internal headers should...
Posted by franz1981 <gi...@git.apache.org>.
Github user franz1981 commented on the issue:
https://github.com/apache/activemq-artemis/pull/1793
@RaiSaurabh Please check the @michaelandrepearce comments about this PR first :+1:
---
[GitHub] activemq-artemis pull request #1793: ARTEMIS-1498: Openwire internal headers...
Posted by michaelandrepearce <gi...@git.apache.org>.
Github user michaelandrepearce commented on a diff in the pull request:
https://github.com/apache/activemq-artemis/pull/1793#discussion_r163007469
--- Diff: artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenwireConverter.java ---
@@ -0,0 +1,45 @@
+/*
+ * 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.activemq.artemis.core.protocol.openwire;
+
+import org.apache.activemq.artemis.api.core.ICoreMessage;
+import org.apache.activemq.artemis.core.message.impl.CoreMessageObjectPools;
+import org.apache.activemq.artemis.spi.core.protocol.MessageConverter;
+
+public class OpenwireConverter implements MessageConverter<OpenwireMessage> {
--- End diff --
nit: But so it fits with other class naming where OpenWire seems to be itself camel cased elsewhere, could this be OpenWireConverter
---
[GitHub] activemq-artemis pull request #1793: ARTEMIS-1498: Openwire internal headers...
Posted by franz1981 <gi...@git.apache.org>.
Github user franz1981 commented on a diff in the pull request:
https://github.com/apache/activemq-artemis/pull/1793#discussion_r165597637
--- Diff: artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireConverter.java ---
@@ -0,0 +1,43 @@
+/*
+ * 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.activemq.artemis.core.protocol.openwire;
+
+import org.apache.activemq.artemis.api.core.ICoreMessage;
+import org.apache.activemq.artemis.core.message.impl.CoreMessageObjectPools;
+import org.apache.activemq.artemis.spi.core.protocol.MessageConverter;
+
+public enum OpenWireConverter implements MessageConverter<OpenWireMessage> {
+
+ INSTANCE;
+
+ public static OpenWireConverter getInstance() {
--- End diff --
This method isn't needed anymore
---
[GitHub] activemq-artemis pull request #1793: ARTEMIS-1498: Openwire internal headers...
Posted by michaelandrepearce <gi...@git.apache.org>.
Github user michaelandrepearce commented on a diff in the pull request:
https://github.com/apache/activemq-artemis/pull/1793#discussion_r165287490
--- Diff: artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireConverter.java ---
@@ -0,0 +1,45 @@
+/*
+ * 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.activemq.artemis.core.protocol.openwire;
+
+import org.apache.activemq.artemis.api.core.ICoreMessage;
+import org.apache.activemq.artemis.core.message.impl.CoreMessageObjectPools;
+import org.apache.activemq.artemis.spi.core.protocol.MessageConverter;
+
+public class OpenWireConverter implements MessageConverter<OpenwireMessage> {
+
+ private static final OpenWireConverter theInstance = new OpenWireConverter();
+ private OpenWireConverter() {
+ }
+
+ public static OpenWireConverter getInstance() {
+ return theInstance;
+ }
+
+ @Override
+ public OpenwireMessage fromCore(ICoreMessage coreMessage) throws Exception {
--- End diff --
seems one is left, i think the class was there before unused, would you mind? OpenwireMessage -> OpenWireMessage
---
[GitHub] activemq-artemis issue #1793: ARTEMIS-1498: Openwire internal headers should...
Posted by michaelandrepearce <gi...@git.apache.org>.
Github user michaelandrepearce commented on the issue:
https://github.com/apache/activemq-artemis/pull/1793
@RaiSaurabh thanks, thats making my life easier, just looking at this atm, if you hadn't guessed :)
---
[GitHub] activemq-artemis pull request #1793: ARTEMIS-1498: Openwire internal headers...
Posted by franz1981 <gi...@git.apache.org>.
Github user franz1981 commented on a diff in the pull request:
https://github.com/apache/activemq-artemis/pull/1793#discussion_r165330750
--- Diff: artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireProtocolManagerFactory.java ---
@@ -46,6 +48,11 @@ public ProtocolManager createProtocolManager(final ActiveMQServer server,
return BeanSupport.setData(new OpenWireProtocolManager(this, server), parameters);
}
+ @Override
+ public Persister<Message>[] getPersister() {
+ return new Persister[]{OpenWireMessagePersister.getInstance()};
--- End diff --
Cache it if possible instead of allocating it each time
---
[GitHub] activemq-artemis issue #1793: ARTEMIS-1498: Openwire internal headers should...
Posted by RaiSaurabh <gi...@git.apache.org>.
Github user RaiSaurabh commented on the issue:
https://github.com/apache/activemq-artemis/pull/1793
@michaelandrepearce updated the code as per your comments.
---
[GitHub] activemq-artemis pull request #1793: ARTEMIS-1498: Openwire internal headers...
Posted by franz1981 <gi...@git.apache.org>.
Github user franz1981 commented on a diff in the pull request:
https://github.com/apache/activemq-artemis/pull/1793#discussion_r165608418
--- Diff: artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireMessageConverter.java ---
@@ -576,308 +334,54 @@ private static ActiveMQMessage toAMQMessage(MessageReference reference,
amqMsg.setExpiration(coreMessage.getExpiration());
amqMsg.setPriority(coreMessage.getPriority());
amqMsg.setTimestamp(coreMessage.getTimestamp());
-
- Long brokerInTime = (Long) coreMessage.getObjectProperty(AMQ_MSG_BROKER_IN_TIME);
- if (brokerInTime == null) {
- brokerInTime = 0L;
- }
- amqMsg.setBrokerInTime(brokerInTime);
-
amqMsg.setCompressed(isCompressed);
- //we need check null because messages may come from other clients
- //and those amq specific attribute may not be set.
- Long arrival = (Long) coreMessage.getObjectProperty(AMQ_MSG_ARRIVAL);
- if (arrival == null) {
- //messages from other sources (like core client) may not set this prop
- arrival = 0L;
- }
- amqMsg.setArrival(arrival);
-
- final String brokerPath = (String) coreMessage.getObjectProperty(AMQ_MSG_BROKER_PATH);
- if (brokerPath != null && !brokerPath.isEmpty()) {
- setAMQMsgBrokerPath(amqMsg, brokerPath);
- }
-
- final String clusterPath = (String) coreMessage.getObjectProperty(AMQ_MSG_CLUSTER);
- if (clusterPath != null && !clusterPath.isEmpty()) {
- setAMQMsgClusterPath(amqMsg, clusterPath);
- }
-
- Integer commandId = (Integer) coreMessage.getObjectProperty(AMQ_MSG_COMMAND_ID);
- if (commandId == null) {
- commandId = -1;
- }
- amqMsg.setCommandId(commandId);
-
final SimpleString corrId = (SimpleString) coreMessage.getObjectProperty(JMS_CORRELATION_ID_PROPERTY);
if (corrId != null) {
amqMsg.setCorrelationId(corrId.toString());
}
- final byte[] dsBytes = (byte[]) coreMessage.getObjectProperty(AMQ_MSG_DATASTRUCTURE);
- if (dsBytes != null) {
- setAMQMsgDataStructure(amqMsg, marshaller, dsBytes);
- }
- final ActiveMQDestination actualDestination = consumer.getOpenwireDestination();
- amqMsg.setDestination(OpenWireUtil.toAMQAddress(coreMessage, actualDestination));
-
- final Object value = coreMessage.getGroupID();
--- End diff --
It is `SimpleString` to could be used the exact type instead of `Object`
---
[GitHub] activemq-artemis pull request #1793: ARTEMIS-1498: Openwire internal headers...
Posted by RaiSaurabh <gi...@git.apache.org>.
Github user RaiSaurabh commented on a diff in the pull request:
https://github.com/apache/activemq-artemis/pull/1793#discussion_r165302018
--- Diff: artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireConverter.java ---
@@ -0,0 +1,45 @@
+/*
+ * 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.activemq.artemis.core.protocol.openwire;
+
+import org.apache.activemq.artemis.api.core.ICoreMessage;
+import org.apache.activemq.artemis.core.message.impl.CoreMessageObjectPools;
+import org.apache.activemq.artemis.spi.core.protocol.MessageConverter;
+
+public class OpenWireConverter implements MessageConverter<OpenwireMessage> {
+
+ private static final OpenWireConverter theInstance = new OpenWireConverter();
+ private OpenWireConverter() {
+ }
+
+ public static OpenWireConverter getInstance() {
+ return theInstance;
+ }
+
+ @Override
+ public OpenwireMessage fromCore(ICoreMessage coreMessage) throws Exception {
--- End diff --
Ok, @michaelandrepearce I will refractor that. Since this was named like this previously I thought not to change it. As after what you suggested this was bugging me too.
---
[GitHub] activemq-artemis pull request #1793: ARTEMIS-1498: Openwire internal headers...
Posted by michaelandrepearce <gi...@git.apache.org>.
Github user michaelandrepearce commented on a diff in the pull request:
https://github.com/apache/activemq-artemis/pull/1793#discussion_r165767867
--- Diff: artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireMessageConverter.java ---
@@ -128,389 +110,159 @@ public Object outbound(org.apache.activemq.artemis.api.core.Message message, int
final WireFormat marshaller,
final CoreMessageObjectPools coreMessageObjectPools) throws Exception {
- final CoreMessage coreMessage = new CoreMessage(-1, messageSend.getSize(), coreMessageObjectPools);
+ final OpenWireMessage openwireMessage = new OpenWireMessage(-1, messageSend.getSize(), coreMessageObjectPools);
final String type = messageSend.getType();
if (type != null) {
- coreMessage.putStringProperty(JMS_TYPE_PROPERTY, new SimpleString(type));
+ openwireMessage.putStringProperty(JMS_TYPE_PROPERTY, new SimpleString(type));
}
- coreMessage.setDurable(messageSend.isPersistent());
- coreMessage.setExpiration(messageSend.getExpiration());
- coreMessage.setPriority(messageSend.getPriority());
- coreMessage.setTimestamp(messageSend.getTimestamp());
-
- final byte coreType = toCoreType(messageSend.getDataStructureType());
- coreMessage.setType(coreType);
+ openwireMessage.setMessageSize(messageSend.getSize());
+ openwireMessage.setDurable(messageSend.isPersistent());
+ openwireMessage.setExpiration(messageSend.getExpiration());
+ openwireMessage.setPriority(messageSend.getPriority());
+ openwireMessage.setTimestamp(messageSend.getTimestamp());
- final ActiveMQBuffer body = coreMessage.getBodyBuffer();
+ final byte coreType = OpenWireCoreConverter.toCoreType(messageSend.getDataStructureType());
+ openwireMessage.setType(coreType);
- final ByteSequence contents = messageSend.getContent();
- if (contents == null && coreType == org.apache.activemq.artemis.api.core.Message.TEXT_TYPE) {
- body.writeNullableString(null);
- } else if (contents != null) {
- final boolean messageCompressed = messageSend.isCompressed();
- if (messageCompressed) {
- coreMessage.putBooleanProperty(AMQ_MSG_COMPRESSED, messageCompressed);
- }
-
- switch (coreType) {
- case org.apache.activemq.artemis.api.core.Message.TEXT_TYPE:
- writeTextType(contents, messageCompressed, body);
- break;
- case org.apache.activemq.artemis.api.core.Message.MAP_TYPE:
- writeMapType(contents, messageCompressed, body);
- break;
- case org.apache.activemq.artemis.api.core.Message.OBJECT_TYPE:
- writeObjectType(contents, messageCompressed, body);
- break;
- case org.apache.activemq.artemis.api.core.Message.STREAM_TYPE:
- writeStreamType(contents, messageCompressed, body);
- break;
- case org.apache.activemq.artemis.api.core.Message.BYTES_TYPE:
- writeBytesType(contents, messageCompressed, body);
- break;
- default:
- writeDefaultType(contents, messageCompressed, body);
- break;
- }
+ final ActiveMQBuffer body = openwireMessage.getBodyBuffer();
+ final boolean messageCompressed = messageSend.isCompressed();
+ if (messageCompressed) {
+ openwireMessage.putBooleanProperty(AMQ_MSG_COMPRESSED, messageCompressed);
}
+ final ByteSequence contents = messageSend.getContent();
+ OpenWireCoreConverter.writeContentIntoBody(body, contents, coreType, messageCompressed);
//amq specific
- coreMessage.putLongProperty(AMQ_MSG_ARRIVAL, messageSend.getArrival());
- coreMessage.putLongProperty(AMQ_MSG_BROKER_IN_TIME, messageSend.getBrokerInTime());
+ openwireMessage.putLongProperty(AMQ_MSG_ARRIVAL, messageSend.getArrival());
+ openwireMessage.putLongProperty(AMQ_MSG_BROKER_IN_TIME, messageSend.getBrokerInTime());
final BrokerId[] brokers = messageSend.getBrokerPath();
if (brokers != null) {
- putMsgBrokerPath(brokers, coreMessage);
+ putMsgBrokerPath(brokers, openwireMessage);
}
final BrokerId[] cluster = messageSend.getCluster();
if (cluster != null) {
- putMsgCluster(cluster, coreMessage);
+ putMsgCluster(cluster, openwireMessage);
}
- coreMessage.putIntProperty(AMQ_MSG_COMMAND_ID, messageSend.getCommandId());
+ openwireMessage.putIntProperty(AMQ_MSG_COMMAND_ID, messageSend.getCommandId());
final String corrId = messageSend.getCorrelationId();
if (corrId != null) {
- coreMessage.putStringProperty(JMS_CORRELATION_ID_PROPERTY, new SimpleString(corrId));
+ openwireMessage.putStringProperty(JMS_CORRELATION_ID_PROPERTY, new SimpleString(corrId));
}
final DataStructure ds = messageSend.getDataStructure();
if (ds != null) {
- putMsgDataStructure(ds, marshaller, coreMessage);
+ putMsgDataStructure(ds, marshaller, openwireMessage);
}
final String groupId = messageSend.getGroupID();
if (groupId != null) {
- coreMessage.putStringProperty(AMQ_MSG_GROUP_ID, new SimpleString(groupId));
+ openwireMessage.putStringProperty(AMQ_MSG_GROUP_ID, new SimpleString(groupId));
--- End diff --
I assume you meant *should
---
[GitHub] activemq-artemis issue #1793: ARTEMIS-1498: Openwire internal headers should...
Posted by michaelandrepearce <gi...@git.apache.org>.
Github user michaelandrepearce commented on the issue:
https://github.com/apache/activemq-artemis/pull/1793
Really good work, I like this, really good to see OpenWire getting this :)
Have you run the full OpenWire test suite? I know not all run in the PR build, so worth checking. Might be worth running the full suite even if you can.
---
[GitHub] activemq-artemis pull request #1793: ARTEMIS-1498: Openwire internal headers...
Posted by franz1981 <gi...@git.apache.org>.
Github user franz1981 commented on a diff in the pull request:
https://github.com/apache/activemq-artemis/pull/1793#discussion_r165606383
--- Diff: artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireMessageConverter.java ---
@@ -128,389 +110,159 @@ public Object outbound(org.apache.activemq.artemis.api.core.Message message, int
final WireFormat marshaller,
final CoreMessageObjectPools coreMessageObjectPools) throws Exception {
- final CoreMessage coreMessage = new CoreMessage(-1, messageSend.getSize(), coreMessageObjectPools);
+ final OpenWireMessage openwireMessage = new OpenWireMessage(-1, messageSend.getSize(), coreMessageObjectPools);
final String type = messageSend.getType();
if (type != null) {
- coreMessage.putStringProperty(JMS_TYPE_PROPERTY, new SimpleString(type));
+ openwireMessage.putStringProperty(JMS_TYPE_PROPERTY, new SimpleString(type));
}
- coreMessage.setDurable(messageSend.isPersistent());
- coreMessage.setExpiration(messageSend.getExpiration());
- coreMessage.setPriority(messageSend.getPriority());
- coreMessage.setTimestamp(messageSend.getTimestamp());
-
- final byte coreType = toCoreType(messageSend.getDataStructureType());
- coreMessage.setType(coreType);
+ openwireMessage.setMessageSize(messageSend.getSize());
+ openwireMessage.setDurable(messageSend.isPersistent());
+ openwireMessage.setExpiration(messageSend.getExpiration());
+ openwireMessage.setPriority(messageSend.getPriority());
+ openwireMessage.setTimestamp(messageSend.getTimestamp());
- final ActiveMQBuffer body = coreMessage.getBodyBuffer();
+ final byte coreType = OpenWireCoreConverter.toCoreType(messageSend.getDataStructureType());
+ openwireMessage.setType(coreType);
- final ByteSequence contents = messageSend.getContent();
- if (contents == null && coreType == org.apache.activemq.artemis.api.core.Message.TEXT_TYPE) {
- body.writeNullableString(null);
- } else if (contents != null) {
- final boolean messageCompressed = messageSend.isCompressed();
- if (messageCompressed) {
- coreMessage.putBooleanProperty(AMQ_MSG_COMPRESSED, messageCompressed);
- }
-
- switch (coreType) {
- case org.apache.activemq.artemis.api.core.Message.TEXT_TYPE:
- writeTextType(contents, messageCompressed, body);
- break;
- case org.apache.activemq.artemis.api.core.Message.MAP_TYPE:
- writeMapType(contents, messageCompressed, body);
- break;
- case org.apache.activemq.artemis.api.core.Message.OBJECT_TYPE:
- writeObjectType(contents, messageCompressed, body);
- break;
- case org.apache.activemq.artemis.api.core.Message.STREAM_TYPE:
- writeStreamType(contents, messageCompressed, body);
- break;
- case org.apache.activemq.artemis.api.core.Message.BYTES_TYPE:
- writeBytesType(contents, messageCompressed, body);
- break;
- default:
- writeDefaultType(contents, messageCompressed, body);
- break;
- }
+ final ActiveMQBuffer body = openwireMessage.getBodyBuffer();
+ final boolean messageCompressed = messageSend.isCompressed();
+ if (messageCompressed) {
+ openwireMessage.putBooleanProperty(AMQ_MSG_COMPRESSED, messageCompressed);
}
+ final ByteSequence contents = messageSend.getContent();
+ OpenWireCoreConverter.writeContentIntoBody(body, contents, coreType, messageCompressed);
//amq specific
- coreMessage.putLongProperty(AMQ_MSG_ARRIVAL, messageSend.getArrival());
- coreMessage.putLongProperty(AMQ_MSG_BROKER_IN_TIME, messageSend.getBrokerInTime());
+ openwireMessage.putLongProperty(AMQ_MSG_ARRIVAL, messageSend.getArrival());
+ openwireMessage.putLongProperty(AMQ_MSG_BROKER_IN_TIME, messageSend.getBrokerInTime());
final BrokerId[] brokers = messageSend.getBrokerPath();
if (brokers != null) {
- putMsgBrokerPath(brokers, coreMessage);
+ putMsgBrokerPath(brokers, openwireMessage);
}
final BrokerId[] cluster = messageSend.getCluster();
if (cluster != null) {
- putMsgCluster(cluster, coreMessage);
+ putMsgCluster(cluster, openwireMessage);
}
- coreMessage.putIntProperty(AMQ_MSG_COMMAND_ID, messageSend.getCommandId());
+ openwireMessage.putIntProperty(AMQ_MSG_COMMAND_ID, messageSend.getCommandId());
final String corrId = messageSend.getCorrelationId();
if (corrId != null) {
- coreMessage.putStringProperty(JMS_CORRELATION_ID_PROPERTY, new SimpleString(corrId));
+ openwireMessage.putStringProperty(JMS_CORRELATION_ID_PROPERTY, new SimpleString(corrId));
}
final DataStructure ds = messageSend.getDataStructure();
if (ds != null) {
- putMsgDataStructure(ds, marshaller, coreMessage);
+ putMsgDataStructure(ds, marshaller, openwireMessage);
}
final String groupId = messageSend.getGroupID();
if (groupId != null) {
- coreMessage.putStringProperty(AMQ_MSG_GROUP_ID, new SimpleString(groupId));
+ openwireMessage.putStringProperty(AMQ_MSG_GROUP_ID, new SimpleString(groupId));
--- End diff --
this could be pooled with `coreMessageObjectPools` as mentioned above
---
[GitHub] activemq-artemis issue #1793: ARTEMIS-1498: Openwire internal headers should...
Posted by clebertsuconic <gi...@git.apache.org>.
Github user clebertsuconic commented on the issue:
https://github.com/apache/activemq-artemis/pull/1793
@michaelandrepearce There is another testsuite just for openwire. somethings that we inherited from activemq5.. it's under ./tests/activemq5-unit-tests
that one was fairly broken. i would need to re-check it.
---
[GitHub] activemq-artemis pull request #1793: ARTEMIS-1498: Openwire internal headers...
Posted by michaelandrepearce <gi...@git.apache.org>.
Github user michaelandrepearce commented on a diff in the pull request:
https://github.com/apache/activemq-artemis/pull/1793#discussion_r165769403
--- Diff: artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireMessageConverter.java ---
@@ -128,389 +110,159 @@ public Object outbound(org.apache.activemq.artemis.api.core.Message message, int
final WireFormat marshaller,
final CoreMessageObjectPools coreMessageObjectPools) throws Exception {
- final CoreMessage coreMessage = new CoreMessage(-1, messageSend.getSize(), coreMessageObjectPools);
+ final OpenWireMessage openwireMessage = new OpenWireMessage(-1, messageSend.getSize(), coreMessageObjectPools);
final String type = messageSend.getType();
if (type != null) {
- coreMessage.putStringProperty(JMS_TYPE_PROPERTY, new SimpleString(type));
+ openwireMessage.putStringProperty(JMS_TYPE_PROPERTY, new SimpleString(type));
}
- coreMessage.setDurable(messageSend.isPersistent());
- coreMessage.setExpiration(messageSend.getExpiration());
- coreMessage.setPriority(messageSend.getPriority());
- coreMessage.setTimestamp(messageSend.getTimestamp());
-
- final byte coreType = toCoreType(messageSend.getDataStructureType());
- coreMessage.setType(coreType);
+ openwireMessage.setMessageSize(messageSend.getSize());
+ openwireMessage.setDurable(messageSend.isPersistent());
+ openwireMessage.setExpiration(messageSend.getExpiration());
+ openwireMessage.setPriority(messageSend.getPriority());
+ openwireMessage.setTimestamp(messageSend.getTimestamp());
- final ActiveMQBuffer body = coreMessage.getBodyBuffer();
+ final byte coreType = OpenWireCoreConverter.toCoreType(messageSend.getDataStructureType());
+ openwireMessage.setType(coreType);
- final ByteSequence contents = messageSend.getContent();
- if (contents == null && coreType == org.apache.activemq.artemis.api.core.Message.TEXT_TYPE) {
- body.writeNullableString(null);
- } else if (contents != null) {
- final boolean messageCompressed = messageSend.isCompressed();
- if (messageCompressed) {
- coreMessage.putBooleanProperty(AMQ_MSG_COMPRESSED, messageCompressed);
- }
-
- switch (coreType) {
- case org.apache.activemq.artemis.api.core.Message.TEXT_TYPE:
- writeTextType(contents, messageCompressed, body);
- break;
- case org.apache.activemq.artemis.api.core.Message.MAP_TYPE:
- writeMapType(contents, messageCompressed, body);
- break;
- case org.apache.activemq.artemis.api.core.Message.OBJECT_TYPE:
- writeObjectType(contents, messageCompressed, body);
- break;
- case org.apache.activemq.artemis.api.core.Message.STREAM_TYPE:
- writeStreamType(contents, messageCompressed, body);
- break;
- case org.apache.activemq.artemis.api.core.Message.BYTES_TYPE:
- writeBytesType(contents, messageCompressed, body);
- break;
- default:
- writeDefaultType(contents, messageCompressed, body);
- break;
- }
+ final ActiveMQBuffer body = openwireMessage.getBodyBuffer();
+ final boolean messageCompressed = messageSend.isCompressed();
+ if (messageCompressed) {
+ openwireMessage.putBooleanProperty(AMQ_MSG_COMPRESSED, messageCompressed);
}
+ final ByteSequence contents = messageSend.getContent();
+ OpenWireCoreConverter.writeContentIntoBody(body, contents, coreType, messageCompressed);
//amq specific
- coreMessage.putLongProperty(AMQ_MSG_ARRIVAL, messageSend.getArrival());
- coreMessage.putLongProperty(AMQ_MSG_BROKER_IN_TIME, messageSend.getBrokerInTime());
+ openwireMessage.putLongProperty(AMQ_MSG_ARRIVAL, messageSend.getArrival());
+ openwireMessage.putLongProperty(AMQ_MSG_BROKER_IN_TIME, messageSend.getBrokerInTime());
final BrokerId[] brokers = messageSend.getBrokerPath();
if (brokers != null) {
- putMsgBrokerPath(brokers, coreMessage);
+ putMsgBrokerPath(brokers, openwireMessage);
}
final BrokerId[] cluster = messageSend.getCluster();
if (cluster != null) {
- putMsgCluster(cluster, coreMessage);
+ putMsgCluster(cluster, openwireMessage);
}
- coreMessage.putIntProperty(AMQ_MSG_COMMAND_ID, messageSend.getCommandId());
+ openwireMessage.putIntProperty(AMQ_MSG_COMMAND_ID, messageSend.getCommandId());
final String corrId = messageSend.getCorrelationId();
if (corrId != null) {
- coreMessage.putStringProperty(JMS_CORRELATION_ID_PROPERTY, new SimpleString(corrId));
+ openwireMessage.putStringProperty(JMS_CORRELATION_ID_PROPERTY, new SimpleString(corrId));
}
final DataStructure ds = messageSend.getDataStructure();
if (ds != null) {
- putMsgDataStructure(ds, marshaller, coreMessage);
+ putMsgDataStructure(ds, marshaller, openwireMessage);
}
final String groupId = messageSend.getGroupID();
if (groupId != null) {
- coreMessage.putStringProperty(AMQ_MSG_GROUP_ID, new SimpleString(groupId));
+ openwireMessage.putStringProperty(AMQ_MSG_GROUP_ID, new SimpleString(groupId));
--- End diff --
as per other comments, id prefer we merge this, and then iterate over the next week, with smaller PR's with these optimisations.
---
[GitHub] activemq-artemis issue #1793: ARTEMIS-1498: Openwire internal headers should...
Posted by michaelandrepearce <gi...@git.apache.org>.
Github user michaelandrepearce commented on the issue:
https://github.com/apache/activemq-artemis/pull/1793
@RaiSaurabh im struggling to rebase this on top of master. Did you rebase after @franz1981's work?
---
[GitHub] activemq-artemis issue #1793: ARTEMIS-1498: Openwire internal headers should...
Posted by RaiSaurabh <gi...@git.apache.org>.
Github user RaiSaurabh commented on the issue:
https://github.com/apache/activemq-artemis/pull/1793
Thanks, @michaelandrepearce @franz1981 for the review. I have implemented the changes.
---
[GitHub] activemq-artemis issue #1793: ARTEMIS-1498: Openwire internal headers should...
Posted by RaiSaurabh <gi...@git.apache.org>.
Github user RaiSaurabh commented on the issue:
https://github.com/apache/activemq-artemis/pull/1793
Seems some issue with Travis and Jenkins for this project is disabled.
---
[GitHub] activemq-artemis pull request #1793: ARTEMIS-1498: Openwire internal headers...
Posted by franz1981 <gi...@git.apache.org>.
Github user franz1981 commented on a diff in the pull request:
https://github.com/apache/activemq-artemis/pull/1793#discussion_r165330940
--- Diff: artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireCoreConverter.java ---
@@ -0,0 +1,572 @@
+/*
+ * 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.activemq.artemis.core.protocol.openwire;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Arrays;
+import java.util.Map;
+import java.util.Set;
+import java.util.Map.Entry;
+import java.util.zip.Deflater;
+import java.util.zip.DeflaterOutputStream;
+import java.util.zip.Inflater;
+import java.util.zip.InflaterInputStream;
+import java.util.zip.InflaterOutputStream;
+
+import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
+import org.apache.activemq.artemis.api.core.FilterConstants;
+import org.apache.activemq.artemis.api.core.ICoreMessage;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.message.impl.CoreMessage;
+import org.apache.activemq.artemis.core.message.impl.CoreMessageObjectPools;
+import org.apache.activemq.artemis.reader.MessageUtil;
+import org.apache.activemq.artemis.utils.DataConstants;
+import org.apache.activemq.artemis.utils.collections.TypedProperties;
+import org.apache.activemq.command.CommandTypes;
+import org.apache.activemq.util.ByteArrayInputStream;
+import org.apache.activemq.util.ByteSequence;
+import org.apache.activemq.util.ByteSequenceData;
+import org.apache.activemq.util.MarshallingSupport;
+import org.fusesource.hawtbuf.UTF8Buffer;
+
+public class OpenWireCoreConverter {
+ public static final SimpleString JMS_GROUPSEQ = new SimpleString("JMSXGroupSequence");
+ public static final SimpleString JMS_XGROUPSEQ = new SimpleString("JMSXGroupSeq");
+ public static final SimpleString JMS_GROUPID = new SimpleString("JMSXGroupID");
+ private static final SimpleString AMQP_REPLYTOGROUPID = new SimpleString("JMS_AMQP_ReplyToGroupID");
+ public static final SimpleString JMS_CONTENT_TYPE = new SimpleString("JMS_AMQP_ContentType");
+ private static final SimpleString AMQP_CONTETTYPE = new SimpleString("JMS_AMQP_CONTENT_TYPE");
+
+ public static ICoreMessage toCore(final OpenWireMessage openwireMessage, final CoreMessageObjectPools coreMessageObjectPools)
+ throws Exception {
+ CoreMessage coreMessage = new CoreMessage(-1, openwireMessage.getMessageSize(), coreMessageObjectPools);
+ final SimpleString type = openwireMessage.getSimpleStringProperty(new SimpleString("JMSType"));
--- End diff --
cache the `new SimpleString("JMSType")` to avoid allocating it each tim
---
[GitHub] activemq-artemis issue #1793: ARTEMIS-1498: Openwire internal headers should...
Posted by clebertsuconic <gi...@git.apache.org>.
Github user clebertsuconic commented on the issue:
https://github.com/apache/activemq-artemis/pull/1793
for this kind of big PR to be merged, we need the full testsuite to be passing.
Not talking about the PR testsuite.
There's an issue that we don't have an env at apache where we can run the whole tests.. so we have to do it on our boxes. I have a box at home where I do this. .and another box at work.
basically:
mvn -Ptests test
and the same on the openwire tests..
---
[GitHub] activemq-artemis pull request #1793: ARTEMIS-1498: Openwire internal headers...
Posted by michaelandrepearce <gi...@git.apache.org>.
Github user michaelandrepearce commented on a diff in the pull request:
https://github.com/apache/activemq-artemis/pull/1793#discussion_r168166465
--- Diff: artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireMessage.java ---
@@ -0,0 +1,673 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.activemq.artemis.core.protocol.openwire;
+
+import javax.jms.JMSException;
+import java.io.IOException;
+import java.util.Enumeration;
+import java.util.HashSet;
+import java.util.Set;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
+import org.apache.activemq.artemis.api.core.ActiveMQException;
+import org.apache.activemq.artemis.api.core.ActiveMQPropertyConversionException;
+import org.apache.activemq.artemis.api.core.ICoreMessage;
+import org.apache.activemq.artemis.api.core.Message;
+import org.apache.activemq.artemis.api.core.RefCountMessage;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.buffers.impl.ChannelBufferWrapper;
+import org.apache.activemq.artemis.core.message.impl.CoreMessageObjectPools;
+import org.apache.activemq.artemis.core.persistence.Persister;
+import org.apache.activemq.artemis.utils.DataConstants;
+import org.apache.activemq.command.ActiveMQDestination;
+import org.apache.activemq.command.ActiveMQMessage;
+import org.apache.activemq.openwire.OpenWireFormat;
+import org.apache.activemq.util.ByteSequence;
+import org.apache.activemq.wireformat.WireFormat;
+
+public class OpenWireMessage extends RefCountMessage {
+
+ private org.apache.activemq.command.ActiveMQMessage message;
+ private WireFormat wireFormat;
+ ByteBuf data;
+ boolean bufferValid;
+
+
+ public OpenWireMessage(org.apache.activemq.command.ActiveMQMessage message, WireFormat wireFormat) {
+ this.message = message;
+ this.wireFormat = wireFormat;
+ try {
+ ByteSequence byteSequence = this.wireFormat.marshal(message);
+ setBuffer(Unpooled.copiedBuffer(byteSequence.getData(), byteSequence.getOffset(), byteSequence.getLength()));
+ } catch (IOException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ public OpenWireMessage() {
+ }
+
+ public void setMarsheller(WireFormat wireFormat) {
+ this.wireFormat = wireFormat;
+ }
+
+ public ActiveMQMessage getAMQMessage() {
+ if (message == null) {
+ if (data != null) {
+ try {
+ message = (ActiveMQMessage) wireFormat.unmarshal(new ChannelBufferWrapper(data));
+ } catch (IOException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return message;
+ } else {
+ return new ActiveMQMessage();
+ }
+ }
+ return message;
+ }
+
+ @Override
+ public void messageChanged() {
+ }
--- End diff --
Does this need implementing
---
[GitHub] activemq-artemis issue #1793: ARTEMIS-1498: Openwire internal headers should...
Posted by RaiSaurabh <gi...@git.apache.org>.
Github user RaiSaurabh commented on the issue:
https://github.com/apache/activemq-artemis/pull/1793
@franz1981 I have implemented the changes @michaelandrepearce requested. @michaelandrepearce could you please review it.
---
[GitHub] activemq-artemis issue #1793: ARTEMIS-1498: Openwire internal headers should...
Posted by michaelandrepearce <gi...@git.apache.org>.
Github user michaelandrepearce commented on the issue:
https://github.com/apache/activemq-artemis/pull/1793
@RaiSaurabh could you rebase your branch on current master then, for me?
---
[GitHub] activemq-artemis issue #1793: ARTEMIS-1498: Openwire internal headers should...
Posted by michaelandrepearce <gi...@git.apache.org>.
Github user michaelandrepearce commented on the issue:
https://github.com/apache/activemq-artemis/pull/1793
i Got a lot of test failures previous, but it seemed a lot better when I ran a suite a few weeks back, is that since his latest changes and work that you ran it? I’m just conscious this would be a really bit +100 on the whole ActiveMQ 6 thing that we all desperately want
---
[GitHub] activemq-artemis pull request #1793: ARTEMIS-1498: Openwire internal headers...
Posted by franz1981 <gi...@git.apache.org>.
Github user franz1981 commented on a diff in the pull request:
https://github.com/apache/activemq-artemis/pull/1793#discussion_r165602190
--- Diff: artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireCoreConverter.java ---
@@ -0,0 +1,572 @@
+/*
+ * 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.activemq.artemis.core.protocol.openwire;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Arrays;
+import java.util.Map;
+import java.util.Set;
+import java.util.Map.Entry;
+import java.util.zip.Deflater;
+import java.util.zip.DeflaterOutputStream;
+import java.util.zip.Inflater;
+import java.util.zip.InflaterInputStream;
+import java.util.zip.InflaterOutputStream;
+
+import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
+import org.apache.activemq.artemis.api.core.FilterConstants;
+import org.apache.activemq.artemis.api.core.ICoreMessage;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.message.impl.CoreMessage;
+import org.apache.activemq.artemis.core.message.impl.CoreMessageObjectPools;
+import org.apache.activemq.artemis.reader.MessageUtil;
+import org.apache.activemq.artemis.utils.DataConstants;
+import org.apache.activemq.artemis.utils.collections.TypedProperties;
+import org.apache.activemq.command.CommandTypes;
+import org.apache.activemq.util.ByteArrayInputStream;
+import org.apache.activemq.util.ByteSequence;
+import org.apache.activemq.util.ByteSequenceData;
+import org.apache.activemq.util.MarshallingSupport;
+import org.fusesource.hawtbuf.UTF8Buffer;
+
+public class OpenWireCoreConverter {
+ public static final SimpleString JMS_GROUPSEQ = new SimpleString("JMSXGroupSequence");
+ public static final SimpleString JMS_XGROUPSEQ = new SimpleString("JMSXGroupSeq");
+ public static final SimpleString JMS_GROUPID = new SimpleString("JMSXGroupID");
+ private static final SimpleString AMQP_REPLYTOGROUPID = new SimpleString("JMS_AMQP_ReplyToGroupID");
+ public static final SimpleString JMS_CONTENT_TYPE = new SimpleString("JMS_AMQP_ContentType");
+ private static final SimpleString AMQP_CONTETTYPE = new SimpleString("JMS_AMQP_CONTENT_TYPE");
+
+ public static ICoreMessage toCore(final OpenWireMessage openwireMessage, final CoreMessageObjectPools coreMessageObjectPools)
+ throws Exception {
+ CoreMessage coreMessage = new CoreMessage(-1, openwireMessage.getMessageSize(), coreMessageObjectPools);
+ final SimpleString type = openwireMessage.getSimpleStringProperty(OpenWireMessageConverter.JMS_TYPE_PROPERTY);
+ if (type != null) {
+ coreMessage.putStringProperty(OpenWireMessageConverter.JMS_TYPE_PROPERTY, type);
+ }
+ coreMessage.setDurable(openwireMessage.isDurable());
+ coreMessage.setExpiration(openwireMessage.getExpiration());
+ coreMessage.setPriority(openwireMessage.getPriority());
+ coreMessage.setTimestamp(openwireMessage.getTimestamp());
+ final ActiveMQBuffer openWireMessageBuffer = openwireMessage.getReadOnlyBodyBuffer();
+ final boolean isCompressed = openwireMessage.getBooleanProperty(OpenWireMessageConverter.AMQ_MSG_COMPRESSED);
+ final byte coreType = openwireMessage.getType();
+ coreMessage.setType(coreType);
+ openWireMessageBuffer.resetReaderIndex();
+ final byte[] bytes;
+ bytes = readContentFromBody(openWireMessageBuffer, coreType, isCompressed);
+ if (bytes != null) {
+ ByteSequence content = new ByteSequence(bytes);
+ ActiveMQBuffer coreMessageBuffer = coreMessage.getBodyBuffer();
+ writeContentIntoBody(coreMessageBuffer, content, coreType, isCompressed);
+ }
+ coreMessage.putObjectProperty(FilterConstants.NATIVE_MESSAGE_ID,
+ openwireMessage.getObjectProperty(FilterConstants.NATIVE_MESSAGE_ID));
+ coreMessage.setMessageID(openwireMessage.getMessageID());
+ final String corrId = openwireMessage.getStringProperty(OpenWireMessageConverter.JMS_CORRELATION_ID_PROPERTY);
+ if (corrId != null) {
+ coreMessage.putStringProperty(OpenWireMessageConverter.JMS_CORRELATION_ID_PROPERTY, corrId);
+ }
+ final String groupId = openwireMessage.getStringProperty(org.apache.activemq.artemis.api.core.Message.HDR_GROUP_ID);
+ if (groupId != null) {
+ coreMessage.putStringProperty(org.apache.activemq.artemis.api.core.Message.HDR_GROUP_ID,
+ new SimpleString(groupId));
+ coreMessage.putStringProperty(JMS_GROUPID, new SimpleString(groupId));
+ }
+ if (openwireMessage.containsProperty(JMS_GROUPSEQ)) {
+ coreMessage.putIntProperty(JMS_GROUPSEQ, openwireMessage.getIntProperty(JMS_GROUPSEQ));
+ coreMessage.putIntProperty(JMS_XGROUPSEQ, openwireMessage.getIntProperty(JMS_GROUPSEQ));
+ }
+ final SimpleString lastValueProperty = openwireMessage.getLastValueProperty();
+ if (lastValueProperty != null) {
+ coreMessage.setLastValueProperty(lastValueProperty);
+ }
+ coreMessage.putObjectProperty(AMQP_CONTETTYPE, openwireMessage.getObjectProperty(JMS_CONTENT_TYPE));
+ coreMessage.putObjectProperty(AMQP_REPLYTOGROUPID, openwireMessage.getObjectProperty(AMQP_REPLYTOGROUPID));
+ coreMessage.setReplyTo(openwireMessage.getReplyTo());
+ coreMessage.setUserID(openwireMessage.getUserID());
+ coreMessage.setDurable(openwireMessage.isDurable());
+ coreMessage.setAddress(openwireMessage.getAddress());
+ coreMessage.putObjectProperty(org.apache.activemq.artemis.api.core.Message.HDR_GROUP_ID,
+ (SimpleString) openwireMessage
+ .getObjectProperty(org.apache.activemq.artemis.api.core.Message.HDR_GROUP_ID));
+ copyAllProperties(openwireMessage, coreMessage);
+ return coreMessage;
+ }
+
+ public static byte toCoreType(byte amqType) {
+ switch (amqType) {
+ case CommandTypes.ACTIVEMQ_BLOB_MESSAGE:
+ throw new IllegalStateException("We don't support BLOB type yet!");
+ case CommandTypes.ACTIVEMQ_BYTES_MESSAGE:
+ return org.apache.activemq.artemis.api.core.Message.BYTES_TYPE;
+ case CommandTypes.ACTIVEMQ_MAP_MESSAGE:
+ return org.apache.activemq.artemis.api.core.Message.MAP_TYPE;
+ case CommandTypes.ACTIVEMQ_OBJECT_MESSAGE:
+ return org.apache.activemq.artemis.api.core.Message.OBJECT_TYPE;
+ case CommandTypes.ACTIVEMQ_STREAM_MESSAGE:
+ return org.apache.activemq.artemis.api.core.Message.STREAM_TYPE;
+ case CommandTypes.ACTIVEMQ_TEXT_MESSAGE:
+ return org.apache.activemq.artemis.api.core.Message.TEXT_TYPE;
+ case CommandTypes.ACTIVEMQ_MESSAGE:
+ return org.apache.activemq.artemis.api.core.Message.DEFAULT_TYPE;
+ default:
+ throw new IllegalStateException("Unknown ActiveMQ Artemis message type: " + amqType);
+ }
+ }
+
+ public static void loadMapIntoProperties(TypedProperties props, Map<String, Object> map) {
+ for (Entry<String, Object> entry : map.entrySet()) {
+ SimpleString key = new SimpleString(entry.getKey());
+ Object value = entry.getValue();
+ if (value instanceof UTF8Buffer) {
+ value = ((UTF8Buffer) value).toString();
--- End diff --
This cast is redundant and could be dropped: it was in the original code, but it won't hurt removing it
---
[GitHub] activemq-artemis issue #1793: ARTEMIS-1498: Openwire internal headers should...
Posted by clebertsuconic <gi...@git.apache.org>.
Github user clebertsuconic commented on the issue:
https://github.com/apache/activemq-artemis/pull/1793
Let me see if I understand.. you implemented OpenwireMessage here?
---
[GitHub] activemq-artemis pull request #1793: ARTEMIS-1498: Openwire internal headers...
Posted by michaelandrepearce <gi...@git.apache.org>.
Github user michaelandrepearce commented on a diff in the pull request:
https://github.com/apache/activemq-artemis/pull/1793#discussion_r165352116
--- Diff: artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireCoreConverter.java ---
@@ -0,0 +1,572 @@
+/*
+ * 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.activemq.artemis.core.protocol.openwire;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Arrays;
+import java.util.Map;
+import java.util.Set;
+import java.util.Map.Entry;
+import java.util.zip.Deflater;
+import java.util.zip.DeflaterOutputStream;
+import java.util.zip.Inflater;
+import java.util.zip.InflaterInputStream;
+import java.util.zip.InflaterOutputStream;
+
+import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
+import org.apache.activemq.artemis.api.core.FilterConstants;
+import org.apache.activemq.artemis.api.core.ICoreMessage;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.message.impl.CoreMessage;
+import org.apache.activemq.artemis.core.message.impl.CoreMessageObjectPools;
+import org.apache.activemq.artemis.reader.MessageUtil;
+import org.apache.activemq.artemis.utils.DataConstants;
+import org.apache.activemq.artemis.utils.collections.TypedProperties;
+import org.apache.activemq.command.CommandTypes;
+import org.apache.activemq.util.ByteArrayInputStream;
+import org.apache.activemq.util.ByteSequence;
+import org.apache.activemq.util.ByteSequenceData;
+import org.apache.activemq.util.MarshallingSupport;
+import org.fusesource.hawtbuf.UTF8Buffer;
+
+public class OpenWireCoreConverter {
+ public static final SimpleString JMS_GROUPSEQ = new SimpleString("JMSXGroupSequence");
+ public static final SimpleString JMS_XGROUPSEQ = new SimpleString("JMSXGroupSeq");
+ public static final SimpleString JMS_GROUPID = new SimpleString("JMSXGroupID");
+ private static final SimpleString AMQP_REPLYTOGROUPID = new SimpleString("JMS_AMQP_ReplyToGroupID");
+ public static final SimpleString JMS_CONTENT_TYPE = new SimpleString("JMS_AMQP_ContentType");
+ private static final SimpleString AMQP_CONTETTYPE = new SimpleString("JMS_AMQP_CONTENT_TYPE");
+
+ public static ICoreMessage toCore(final OpenWireMessage openwireMessage, final CoreMessageObjectPools coreMessageObjectPools)
+ throws Exception {
+ CoreMessage coreMessage = new CoreMessage(-1, openwireMessage.getMessageSize(), coreMessageObjectPools);
+ final SimpleString type = openwireMessage.getSimpleStringProperty(new SimpleString("JMSType"));
--- End diff --
Or make constant :)
---
[GitHub] activemq-artemis pull request #1793: ARTEMIS-1498: Openwire internal headers...
Posted by franz1981 <gi...@git.apache.org>.
Github user franz1981 commented on a diff in the pull request:
https://github.com/apache/activemq-artemis/pull/1793#discussion_r165601509
--- Diff: artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireCoreConverter.java ---
@@ -0,0 +1,572 @@
+/*
+ * 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.activemq.artemis.core.protocol.openwire;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Arrays;
+import java.util.Map;
+import java.util.Set;
+import java.util.Map.Entry;
+import java.util.zip.Deflater;
+import java.util.zip.DeflaterOutputStream;
+import java.util.zip.Inflater;
+import java.util.zip.InflaterInputStream;
+import java.util.zip.InflaterOutputStream;
+
+import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
+import org.apache.activemq.artemis.api.core.FilterConstants;
+import org.apache.activemq.artemis.api.core.ICoreMessage;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.message.impl.CoreMessage;
+import org.apache.activemq.artemis.core.message.impl.CoreMessageObjectPools;
+import org.apache.activemq.artemis.reader.MessageUtil;
+import org.apache.activemq.artemis.utils.DataConstants;
+import org.apache.activemq.artemis.utils.collections.TypedProperties;
+import org.apache.activemq.command.CommandTypes;
+import org.apache.activemq.util.ByteArrayInputStream;
+import org.apache.activemq.util.ByteSequence;
+import org.apache.activemq.util.ByteSequenceData;
+import org.apache.activemq.util.MarshallingSupport;
+import org.fusesource.hawtbuf.UTF8Buffer;
+
+public class OpenWireCoreConverter {
+ public static final SimpleString JMS_GROUPSEQ = new SimpleString("JMSXGroupSequence");
+ public static final SimpleString JMS_XGROUPSEQ = new SimpleString("JMSXGroupSeq");
+ public static final SimpleString JMS_GROUPID = new SimpleString("JMSXGroupID");
+ private static final SimpleString AMQP_REPLYTOGROUPID = new SimpleString("JMS_AMQP_ReplyToGroupID");
+ public static final SimpleString JMS_CONTENT_TYPE = new SimpleString("JMS_AMQP_ContentType");
+ private static final SimpleString AMQP_CONTETTYPE = new SimpleString("JMS_AMQP_CONTENT_TYPE");
+
+ public static ICoreMessage toCore(final OpenWireMessage openwireMessage, final CoreMessageObjectPools coreMessageObjectPools)
+ throws Exception {
+ CoreMessage coreMessage = new CoreMessage(-1, openwireMessage.getMessageSize(), coreMessageObjectPools);
+ final SimpleString type = openwireMessage.getSimpleStringProperty(OpenWireMessageConverter.JMS_TYPE_PROPERTY);
+ if (type != null) {
+ coreMessage.putStringProperty(OpenWireMessageConverter.JMS_TYPE_PROPERTY, type);
+ }
+ coreMessage.setDurable(openwireMessage.isDurable());
+ coreMessage.setExpiration(openwireMessage.getExpiration());
+ coreMessage.setPriority(openwireMessage.getPriority());
+ coreMessage.setTimestamp(openwireMessage.getTimestamp());
+ final ActiveMQBuffer openWireMessageBuffer = openwireMessage.getReadOnlyBodyBuffer();
+ final boolean isCompressed = openwireMessage.getBooleanProperty(OpenWireMessageConverter.AMQ_MSG_COMPRESSED);
+ final byte coreType = openwireMessage.getType();
+ coreMessage.setType(coreType);
+ openWireMessageBuffer.resetReaderIndex();
+ final byte[] bytes;
+ bytes = readContentFromBody(openWireMessageBuffer, coreType, isCompressed);
+ if (bytes != null) {
+ ByteSequence content = new ByteSequence(bytes);
+ ActiveMQBuffer coreMessageBuffer = coreMessage.getBodyBuffer();
+ writeContentIntoBody(coreMessageBuffer, content, coreType, isCompressed);
+ }
+ coreMessage.putObjectProperty(FilterConstants.NATIVE_MESSAGE_ID,
+ openwireMessage.getObjectProperty(FilterConstants.NATIVE_MESSAGE_ID));
+ coreMessage.setMessageID(openwireMessage.getMessageID());
+ final String corrId = openwireMessage.getStringProperty(OpenWireMessageConverter.JMS_CORRELATION_ID_PROPERTY);
+ if (corrId != null) {
+ coreMessage.putStringProperty(OpenWireMessageConverter.JMS_CORRELATION_ID_PROPERTY, corrId);
+ }
+ final String groupId = openwireMessage.getStringProperty(org.apache.activemq.artemis.api.core.Message.HDR_GROUP_ID);
+ if (groupId != null) {
+ coreMessage.putStringProperty(org.apache.activemq.artemis.api.core.Message.HDR_GROUP_ID,
+ new SimpleString(groupId));
+ coreMessage.putStringProperty(JMS_GROUPID, new SimpleString(groupId));
+ }
+ if (openwireMessage.containsProperty(JMS_GROUPSEQ)) {
+ coreMessage.putIntProperty(JMS_GROUPSEQ, openwireMessage.getIntProperty(JMS_GROUPSEQ));
+ coreMessage.putIntProperty(JMS_XGROUPSEQ, openwireMessage.getIntProperty(JMS_GROUPSEQ));
+ }
+ final SimpleString lastValueProperty = openwireMessage.getLastValueProperty();
+ if (lastValueProperty != null) {
+ coreMessage.setLastValueProperty(lastValueProperty);
+ }
+ coreMessage.putObjectProperty(AMQP_CONTETTYPE, openwireMessage.getObjectProperty(JMS_CONTENT_TYPE));
+ coreMessage.putObjectProperty(AMQP_REPLYTOGROUPID, openwireMessage.getObjectProperty(AMQP_REPLYTOGROUPID));
+ coreMessage.setReplyTo(openwireMessage.getReplyTo());
+ coreMessage.setUserID(openwireMessage.getUserID());
+ coreMessage.setDurable(openwireMessage.isDurable());
+ coreMessage.setAddress(openwireMessage.getAddress());
+ coreMessage.putObjectProperty(org.apache.activemq.artemis.api.core.Message.HDR_GROUP_ID,
--- End diff --
it isn't happening twice? Some lines above there is a similar thing:
```
final String groupId = openwireMessage.getStringProperty(org.apache.activemq.artemis.api.core.Message.HDR_GROUP_ID);
if (groupId != null) {
coreMessage.putStringProperty(org.apache.activemq.artemis.api.core.Message.HDR_GROUP_ID,
new SimpleString(groupId));
```
---
[GitHub] activemq-artemis issue #1793: ARTEMIS-1498: Openwire internal headers should...
Posted by clebertsuconic <gi...@git.apache.org>.
Github user clebertsuconic commented on the issue:
https://github.com/apache/activemq-artemis/pull/1793
this is a nice start.. but it will require some work... a lot of tests are broken... the converters need to be updated... (especially on the openwire testsuite).
---
[GitHub] activemq-artemis pull request #1793: ARTEMIS-1498: Openwire internal headers...
Posted by franz1981 <gi...@git.apache.org>.
Github user franz1981 commented on a diff in the pull request:
https://github.com/apache/activemq-artemis/pull/1793#discussion_r165603319
--- Diff: artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireMessage.java ---
@@ -0,0 +1,1152 @@
+/**
+ * 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.activemq.artemis.core.protocol.openwire;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.util.LinkedList;
+import java.util.Set;
+
+import io.netty.buffer.ByteBuf;
+import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
+import org.apache.activemq.artemis.api.core.ActiveMQBuffers;
+import org.apache.activemq.artemis.api.core.ActiveMQException;
+import org.apache.activemq.artemis.api.core.ActiveMQPropertyConversionException;
+import org.apache.activemq.artemis.api.core.ICoreMessage;
+import org.apache.activemq.artemis.api.core.Message;
+import org.apache.activemq.artemis.api.core.RefCountMessage;
+import org.apache.activemq.artemis.api.core.RoutingType;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.buffers.impl.ChannelBufferWrapper;
+import org.apache.activemq.artemis.core.buffers.impl.ResetLimitWrappedActiveMQBuffer;
+import org.apache.activemq.artemis.core.message.LargeBodyEncoder;
+import org.apache.activemq.artemis.core.message.impl.CoreMessageObjectPools;
+import org.apache.activemq.artemis.core.persistence.Persister;
+import org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl;
+import org.apache.activemq.artemis.reader.MessageUtil;
+import org.apache.activemq.artemis.utils.DataConstants;
+import org.apache.activemq.artemis.utils.UUID;
+import org.apache.activemq.artemis.utils.collections.TypedProperties;
+import org.jboss.logging.Logger;
+
+/** Note: you shouldn't change properties using multi-threads. Change your properties before you can send it to multiple
+ * consumers */
+public class OpenWireMessage extends RefCountMessage {
--- End diff --
optional: If it is not meant to be subclassed makes it `final` and let `protected` fields become `private` to reduce visibility
---
[GitHub] activemq-artemis issue #1793: ARTEMIS-1498: Openwire internal headers should...
Posted by jbertram <gi...@git.apache.org>.
Github user jbertram commented on the issue:
https://github.com/apache/activemq-artemis/pull/1793
@RaiSaurabh, at this point there are lots of conflicts on this PR. They either need to be fixed and this PR can move forward or it needs to be closed.
---
[GitHub] activemq-artemis issue #1793: ARTEMIS-1498: Openwire internal headers should...
Posted by RaiSaurabh <gi...@git.apache.org>.
Github user RaiSaurabh commented on the issue:
https://github.com/apache/activemq-artemis/pull/1793
Thanks, @michaelandrepearce @clebertsuconic and @franz1981 for the review comments. I have tried to fix it. Could you please review the changes now.
---
[GitHub] activemq-artemis issue #1793: ARTEMIS-1498: Openwire internal headers should...
Posted by michaelandrepearce <gi...@git.apache.org>.
Github user michaelandrepearce commented on the issue:
https://github.com/apache/activemq-artemis/pull/1793
@RaiSaurabh have sent a PR to your branch, that solves some bits for you.
---
[GitHub] activemq-artemis issue #1793: ARTEMIS-1498: Openwire internal headers should...
Posted by RaiSaurabh <gi...@git.apache.org>.
Github user RaiSaurabh commented on the issue:
https://github.com/apache/activemq-artemis/pull/1793
@michaelandrepearce Yes, I did.
---
[GitHub] activemq-artemis issue #1793: ARTEMIS-1498: Openwire internal headers should...
Posted by clebertsuconic <gi...@git.apache.org>.
Github user clebertsuconic commented on the issue:
https://github.com/apache/activemq-artemis/pull/1793
There should be some simplification on the converters here. They shouldn't be needed at all when talking openwire <-> openwire. I saw some issues when I was looking into this.
---
[GitHub] activemq-artemis pull request #1793: ARTEMIS-1498: Openwire internal headers...
Posted by michaelandrepearce <gi...@git.apache.org>.
Github user michaelandrepearce commented on a diff in the pull request:
https://github.com/apache/activemq-artemis/pull/1793#discussion_r168166559
--- Diff: artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireMessage.java ---
@@ -0,0 +1,673 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.activemq.artemis.core.protocol.openwire;
+
+import javax.jms.JMSException;
+import java.io.IOException;
+import java.util.Enumeration;
+import java.util.HashSet;
+import java.util.Set;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
+import org.apache.activemq.artemis.api.core.ActiveMQException;
+import org.apache.activemq.artemis.api.core.ActiveMQPropertyConversionException;
+import org.apache.activemq.artemis.api.core.ICoreMessage;
+import org.apache.activemq.artemis.api.core.Message;
+import org.apache.activemq.artemis.api.core.RefCountMessage;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.buffers.impl.ChannelBufferWrapper;
+import org.apache.activemq.artemis.core.message.impl.CoreMessageObjectPools;
+import org.apache.activemq.artemis.core.persistence.Persister;
+import org.apache.activemq.artemis.utils.DataConstants;
+import org.apache.activemq.command.ActiveMQDestination;
+import org.apache.activemq.command.ActiveMQMessage;
+import org.apache.activemq.openwire.OpenWireFormat;
+import org.apache.activemq.util.ByteSequence;
+import org.apache.activemq.wireformat.WireFormat;
+
+public class OpenWireMessage extends RefCountMessage {
+
+ private org.apache.activemq.command.ActiveMQMessage message;
+ private WireFormat wireFormat;
+ ByteBuf data;
+ boolean bufferValid;
+
+
+ public OpenWireMessage(org.apache.activemq.command.ActiveMQMessage message, WireFormat wireFormat) {
+ this.message = message;
+ this.wireFormat = wireFormat;
+ try {
+ ByteSequence byteSequence = this.wireFormat.marshal(message);
+ setBuffer(Unpooled.copiedBuffer(byteSequence.getData(), byteSequence.getOffset(), byteSequence.getLength()));
+ } catch (IOException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ public OpenWireMessage() {
+ }
+
+ public void setMarsheller(WireFormat wireFormat) {
+ this.wireFormat = wireFormat;
+ }
+
+ public ActiveMQMessage getAMQMessage() {
+ if (message == null) {
+ if (data != null) {
+ try {
+ message = (ActiveMQMessage) wireFormat.unmarshal(new ChannelBufferWrapper(data));
+ } catch (IOException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return message;
+ } else {
+ return new ActiveMQMessage();
+ }
+ }
+ return message;
+ }
+
+ @Override
+ public void messageChanged() {
+ }
+
+ @Override
+ public Long getScheduledDeliveryTime() {
+ return message.getArrival();
+ }
+
+ @Override
+ public SimpleString getReplyTo() {
+ return SimpleString.toSimpleString(message.getReplyTo().getPhysicalName());
--- End diff --
Could use pool
---
[GitHub] activemq-artemis pull request #1793: ARTEMIS-1498: Openwire internal headers...
Posted by franz1981 <gi...@git.apache.org>.
Github user franz1981 commented on a diff in the pull request:
https://github.com/apache/activemq-artemis/pull/1793#discussion_r165331112
--- Diff: artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireCoreConverter.java ---
@@ -0,0 +1,572 @@
+/*
+ * 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.activemq.artemis.core.protocol.openwire;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Arrays;
+import java.util.Map;
+import java.util.Set;
+import java.util.Map.Entry;
+import java.util.zip.Deflater;
+import java.util.zip.DeflaterOutputStream;
+import java.util.zip.Inflater;
+import java.util.zip.InflaterInputStream;
+import java.util.zip.InflaterOutputStream;
+
+import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
+import org.apache.activemq.artemis.api.core.FilterConstants;
+import org.apache.activemq.artemis.api.core.ICoreMessage;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.message.impl.CoreMessage;
+import org.apache.activemq.artemis.core.message.impl.CoreMessageObjectPools;
+import org.apache.activemq.artemis.reader.MessageUtil;
+import org.apache.activemq.artemis.utils.DataConstants;
+import org.apache.activemq.artemis.utils.collections.TypedProperties;
+import org.apache.activemq.command.CommandTypes;
+import org.apache.activemq.util.ByteArrayInputStream;
+import org.apache.activemq.util.ByteSequence;
+import org.apache.activemq.util.ByteSequenceData;
+import org.apache.activemq.util.MarshallingSupport;
+import org.fusesource.hawtbuf.UTF8Buffer;
+
+public class OpenWireCoreConverter {
+ public static final SimpleString JMS_GROUPSEQ = new SimpleString("JMSXGroupSequence");
+ public static final SimpleString JMS_XGROUPSEQ = new SimpleString("JMSXGroupSeq");
+ public static final SimpleString JMS_GROUPID = new SimpleString("JMSXGroupID");
+ private static final SimpleString AMQP_REPLYTOGROUPID = new SimpleString("JMS_AMQP_ReplyToGroupID");
+ public static final SimpleString JMS_CONTENT_TYPE = new SimpleString("JMS_AMQP_ContentType");
+ private static final SimpleString AMQP_CONTETTYPE = new SimpleString("JMS_AMQP_CONTENT_TYPE");
+
+ public static ICoreMessage toCore(final OpenWireMessage openwireMessage, final CoreMessageObjectPools coreMessageObjectPools)
+ throws Exception {
+ CoreMessage coreMessage = new CoreMessage(-1, openwireMessage.getMessageSize(), coreMessageObjectPools);
+ final SimpleString type = openwireMessage.getSimpleStringProperty(new SimpleString("JMSType"));
+ if (type != null) {
+ coreMessage.putStringProperty(SimpleString.toSimpleString("JMSType"), type);
--- End diff --
Cache `SimpleString.toSimpleString("JMSType")` to avoid allocations
---
[GitHub] activemq-artemis issue #1793: ARTEMIS-1498: Openwire internal headers should...
Posted by franz1981 <gi...@git.apache.org>.
Github user franz1981 commented on the issue:
https://github.com/apache/activemq-artemis/pull/1793
Please take a look here:
```
2018-01-22 17:28:59,899 WARN [org.apache.activemq.artemis.core.server] Error during message dispatch: java.lang.RuntimeException: null
at org.apache.activemq.artemis.core.protocol.openwire.OpenwireMessage.toCore(OpenwireMessage.java:787) [artemis-openwire-protocol-2.5.0-SNAPSHOT.jar:]
at org.apache.activemq.artemis.core.protocol.openwire.OpenwireMessage.toCore(OpenwireMessage.java:779) [artemis-openwire-protocol-2.5.0-SNAPSHOT.jar:]
at org.apache.activemq.artemis.core.protocol.openwire.OpenWireMessageConverter.toAMQMessage(OpenWireMessageConverter.java:454) [artemis-openwire-protocol-2.5.0-SNAPSHOT.jar:]
at org.apache.activemq.artemis.core.protocol.openwire.OpenWireMessageConverter.createMessageDispatch(OpenWireMessageConverter.java:435) [artemis-openwire-protocol-2.5.0-SNAPSHOT.jar:]
at org.apache.activemq.artemis.core.protocol.openwire.amq.AMQConsumer.handleDeliver(AMQConsumer.java:222) [artemis-openwire-protocol-2.5.0-SNAPSHOT.jar:]
at org.apache.activemq.artemis.core.protocol.openwire.amq.AMQSession.sendMessage(AMQSession.java:276) [artemis-openwire-protocol-2.5.0-SNAPSHOT.jar:]
at org.apache.activemq.artemis.core.server.impl.ServerConsumerImpl.deliverStandardMessage(ServerConsumerImpl.java:1091) [artemis-server-2.5.0-SNAPSHOT.jar:2.5.0-SNAPSHOT]
at org.apache.activemq.artemis.core.server.impl.ServerConsumerImpl.proceedDeliver(ServerConsumerImpl.java:460) [artemis-server-2.5.0-SNAPSHOT.jar:2.5.0-SNAPSHOT]
at org.apache.activemq.artemis.core.server.impl.QueueImpl.proceedDeliver(QueueImpl.java:2763) [artemis-server-2.5.0-SNAPSHOT.jar:2.5.0-SNAPSHOT]
at org.apache.activemq.artemis.core.server.impl.QueueImpl.deliver(QueueImpl.java:2247) [artemis-server-2.5.0-SNAPSHOT.jar:2.5.0-SNAPSHOT]
at org.apache.activemq.artemis.core.server.impl.QueueImpl.access$1900(QueueImpl.java:106) [artemis-server-2.5.0-SNAPSHOT.jar:2.5.0-SNAPSHOT]
at org.apache.activemq.artemis.core.server.impl.QueueImpl$DeliverRunner.run(QueueImpl.java:3021) [artemis-server-2.5.0-SNAPSHOT.jar:2.5.0-SNAPSHOT]
at org.apache.activemq.artemis.utils.actors.OrderedExecutor.doTask(OrderedExecutor.java:42) [artemis-commons-2.5.0-SNAPSHOT.jar:2.5.0-SNAPSHOT]
at org.apache.activemq.artemis.utils.actors.OrderedExecutor.doTask(OrderedExecutor.java:31) [artemis-commons-2.5.0-SNAPSHOT.jar:2.5.0-SNAPSHOT]
at org.apache.activemq.artemis.utils.actors.ProcessorBase.executePendingTasks(ProcessorBase.java:66) [artemis-commons-2.5.0-SNAPSHOT.jar:2.5.0-SNAPSHOT]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) [rt.jar:1.8.0_102]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) [rt.jar:1.8.0_102]
at java.lang.Thread.run(Thread.java:745) [rt.jar:1.8.0_102]
Caused by: java.lang.NumberFormatException: null
at java.lang.Integer.parseInt(Integer.java:542) [rt.jar:1.8.0_102]
at java.lang.Integer.valueOf(Integer.java:766) [rt.jar:1.8.0_102]
at org.apache.activemq.artemis.utils.collections.TypedProperties.getIntProperty(TypedProperties.java:225) [artemis-commons-2.5.0-SNAPSHOT.jar:2.5.0-SNAPSHOT]
at org.apache.activemq.artemis.core.protocol.openwire.OpenwireMessage.getIntProperty(OpenwireMessage.java:681) [artemis-openwire-protocol-2.5.0-SNAPSHOT.jar:]
at org.apache.activemq.artemis.core.protocol.openwire.OpenwireMessage.getIntProperty(OpenwireMessage.java:600) [artemis-openwire-protocol-2.5.0-SNAPSHOT.jar:]
at org.apache.activemq.artemis.core.protocol.openwire.OpenwireCoreConverter.toCore(OpenwireCoreConverter.java:74) [artemis-openwire-protocol-2.5.0-SNAPSHOT.jar:]
at org.apache.activemq.artemis.core.protocol.openwire.OpenwireConverter.toCore(OpenwireConverter.java:42) [artemis-openwire-protocol-2.5.0-SNAPSHOT.jar:]
at org.apache.activemq.artemis.core.protocol.openwire.OpenwireMessage.toCore(OpenwireMessage.java:785) [artemis-openwire-protocol-2.5.0-SNAPSHOT.jar:]
... 17 more
```
I've built up a 1 producer 1 consumer test vs 1 not durable JMS queue and I'm getting it.
The idea seems good as @clebertsuconic as shown me but need to address these issues first :+1:
---
[GitHub] activemq-artemis issue #1793: ARTEMIS-1498: Openwire internal headers should...
Posted by franz1981 <gi...@git.apache.org>.
Github user franz1981 commented on the issue:
https://github.com/apache/activemq-artemis/pull/1793
@RaiSaurabh I've just checked vs the `integration/openwire` tests in `integration-tests` and it fail 22 tests/338.
The original number of failing tests here is 1, so anything over this value is actually a regression.
I'm not runining the compatibility ones on `activemq5-unit-tests` (my notebook could die before finishing them :P) with no reasons.
Please run the tests I've mentioned and try at least to no have regressions there: the idea of the original PR is good (very good), but the implementation need to be fixed in order to not have huge regressions on how OpenWire is working, thanks :+1:
---
[GitHub] activemq-artemis pull request #1793: ARTEMIS-1498: Openwire internal headers...
Posted by franz1981 <gi...@git.apache.org>.
Github user franz1981 commented on a diff in the pull request:
https://github.com/apache/activemq-artemis/pull/1793#discussion_r165330188
--- Diff: artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireConverter.java ---
@@ -0,0 +1,45 @@
+/*
+ * 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.activemq.artemis.core.protocol.openwire;
+
+import org.apache.activemq.artemis.api.core.ICoreMessage;
+import org.apache.activemq.artemis.core.message.impl.CoreMessageObjectPools;
+import org.apache.activemq.artemis.spi.core.protocol.MessageConverter;
+
+public class OpenWireConverter implements MessageConverter<OpenWireMessage> {
--- End diff --
Please make it final or use an enum class with just a single enumerated instance to have a real singleton:
```
public enum OpenWireConverter implements MessageConverter<OpenWireMessage>{
Instance;
...
}
```
---
[GitHub] activemq-artemis issue #1793: ARTEMIS-1498: Openwire internal headers should...
Posted by clebertsuconic <gi...@git.apache.org>.
Github user clebertsuconic commented on the issue:
https://github.com/apache/activemq-artemis/pull/1793
I ran the whole testsuite and it didn't complete.
it's a very nice start though.
---
[GitHub] activemq-artemis pull request #1793: ARTEMIS-1498: Openwire internal headers...
Posted by franz1981 <gi...@git.apache.org>.
Github user franz1981 commented on a diff in the pull request:
https://github.com/apache/activemq-artemis/pull/1793#discussion_r165599007
--- Diff: artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireCoreConverter.java ---
@@ -0,0 +1,572 @@
+/*
+ * 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.activemq.artemis.core.protocol.openwire;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Arrays;
+import java.util.Map;
+import java.util.Set;
+import java.util.Map.Entry;
+import java.util.zip.Deflater;
+import java.util.zip.DeflaterOutputStream;
+import java.util.zip.Inflater;
+import java.util.zip.InflaterInputStream;
+import java.util.zip.InflaterOutputStream;
+
+import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
+import org.apache.activemq.artemis.api.core.FilterConstants;
+import org.apache.activemq.artemis.api.core.ICoreMessage;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.message.impl.CoreMessage;
+import org.apache.activemq.artemis.core.message.impl.CoreMessageObjectPools;
+import org.apache.activemq.artemis.reader.MessageUtil;
+import org.apache.activemq.artemis.utils.DataConstants;
+import org.apache.activemq.artemis.utils.collections.TypedProperties;
+import org.apache.activemq.command.CommandTypes;
+import org.apache.activemq.util.ByteArrayInputStream;
+import org.apache.activemq.util.ByteSequence;
+import org.apache.activemq.util.ByteSequenceData;
+import org.apache.activemq.util.MarshallingSupport;
+import org.fusesource.hawtbuf.UTF8Buffer;
+
+public class OpenWireCoreConverter {
--- End diff --
It is optional: If it won't be instantiated anymore, please make it `final` and put a private constructor on it in order to avoid "accidental" instantiations of it
---
[GitHub] activemq-artemis issue #1793: ARTEMIS-1498: Openwire internal headers should...
Posted by michaelandrepearce <gi...@git.apache.org>.
Github user michaelandrepearce commented on the issue:
https://github.com/apache/activemq-artemis/pull/1793
Really need PR build to be green. Maybe re-push with a comment update to get try trigger another build.
Seems some methods have been still left in-implemented. Also could/should look to use string to SimpleString pools.
Lastly have you checked that this persists to disk and is recoverable without converting to core.
---
[GitHub] activemq-artemis issue #1793: ARTEMIS-1498: Openwire internal headers should...
Posted by franz1981 <gi...@git.apache.org>.
Github user franz1981 commented on the issue:
https://github.com/apache/activemq-artemis/pull/1793
@michaelandrepearce I will warn the PM squad :+1:
---
[GitHub] activemq-artemis issue #1793: ARTEMIS-1498: Openwire internal headers should...
Posted by franz1981 <gi...@git.apache.org>.
Github user franz1981 commented on the issue:
https://github.com/apache/activemq-artemis/pull/1793
@RaiSaurabh @michaelandrepearce
I've run a couple of benchmarks and the results are strange, perf wire the improvements are on par or (slightly) less than https://github.com/apache/activemq-artemis/pull/1849, but the number are "strange" eg:
```
4 producers - 4 consumers
------------------------------------
EndToEnd Throughput: 145985 ops/sec
EndToEnd SERVICE-TIME Latencies distribution in MICROSECONDS
mean 10032775.17
min 9999220.74
50.00% 10066329.60
90.00% 10066329.60
99.00% 10066329.60
99.90% 10066329.60
99.99% 10066329.60
max 10066329.60
count 400000
---------------------------------
1 producer - 1 consumer
---------------------------------
EndToEnd Throughput: 78247 ops/sec
EndToEnd SERVICE-TIME Latencies distribution in MICROSECONDS
mean 10032775.17
min 9999220.74
50.00% 10066329.60
90.00% 10066329.60
99.00% 10066329.60
99.90% 10066329.60
99.99% 10066329.60
max 10066329.60
count 100000
```
It is weird: the end-2-end latencies are stucked being ALWAYS the same with performances a little less from the PR above: probably there is something not working on message translation, please takes a look to the Open Wire tests to spot if everything is ok...
---
[GitHub] activemq-artemis pull request #1793: ARTEMIS-1498: Openwire internal headers...
Posted by michaelandrepearce <gi...@git.apache.org>.
Github user michaelandrepearce commented on a diff in the pull request:
https://github.com/apache/activemq-artemis/pull/1793#discussion_r168166944
--- Diff: artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireMessage.java ---
@@ -0,0 +1,673 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.activemq.artemis.core.protocol.openwire;
+
+import javax.jms.JMSException;
+import java.io.IOException;
+import java.util.Enumeration;
+import java.util.HashSet;
+import java.util.Set;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
+import org.apache.activemq.artemis.api.core.ActiveMQException;
+import org.apache.activemq.artemis.api.core.ActiveMQPropertyConversionException;
+import org.apache.activemq.artemis.api.core.ICoreMessage;
+import org.apache.activemq.artemis.api.core.Message;
+import org.apache.activemq.artemis.api.core.RefCountMessage;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.buffers.impl.ChannelBufferWrapper;
+import org.apache.activemq.artemis.core.message.impl.CoreMessageObjectPools;
+import org.apache.activemq.artemis.core.persistence.Persister;
+import org.apache.activemq.artemis.utils.DataConstants;
+import org.apache.activemq.command.ActiveMQDestination;
+import org.apache.activemq.command.ActiveMQMessage;
+import org.apache.activemq.openwire.OpenWireFormat;
+import org.apache.activemq.util.ByteSequence;
+import org.apache.activemq.wireformat.WireFormat;
+
+public class OpenWireMessage extends RefCountMessage {
+
+ private org.apache.activemq.command.ActiveMQMessage message;
+ private WireFormat wireFormat;
+ ByteBuf data;
+ boolean bufferValid;
+
+
+ public OpenWireMessage(org.apache.activemq.command.ActiveMQMessage message, WireFormat wireFormat) {
+ this.message = message;
+ this.wireFormat = wireFormat;
+ try {
+ ByteSequence byteSequence = this.wireFormat.marshal(message);
+ setBuffer(Unpooled.copiedBuffer(byteSequence.getData(), byteSequence.getOffset(), byteSequence.getLength()));
+ } catch (IOException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ public OpenWireMessage() {
+ }
+
+ public void setMarsheller(WireFormat wireFormat) {
+ this.wireFormat = wireFormat;
+ }
+
+ public ActiveMQMessage getAMQMessage() {
+ if (message == null) {
+ if (data != null) {
+ try {
+ message = (ActiveMQMessage) wireFormat.unmarshal(new ChannelBufferWrapper(data));
+ } catch (IOException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return message;
+ } else {
+ return new ActiveMQMessage();
+ }
+ }
+ return message;
+ }
+
+ @Override
+ public void messageChanged() {
+ }
+
+ @Override
+ public Long getScheduledDeliveryTime() {
+ return message.getArrival();
+ }
+
+ @Override
+ public SimpleString getReplyTo() {
+ return SimpleString.toSimpleString(message.getReplyTo().getPhysicalName());
+ }
+
+ @Override
+ public Message setReplyTo(SimpleString address) {
+ message.setReplyTo(ActiveMQDestination.createDestination(address.toString(), ActiveMQDestination.QUEUE_TYPE));
+ return this;
+ }
+
+ @Override
+ public Message setBuffer(ByteBuf buffer) {
+ this.data = buffer;
+ return this;
+ }
+
+ @Override
+ public ByteBuf getBuffer() {
+ if (data == null) {
+ return null;
+ } else {
+ return Unpooled.wrappedBuffer(data);
+ }
+ }
+
+ @Override
+ public OpenWireMessage copy() {
+ return new OpenWireMessage((ActiveMQMessage)message.copy(), wireFormat);
+ }
+
+ @Override
+ public Message copy(long newID) {
+ OpenWireMessage copy = new OpenWireMessage((ActiveMQMessage)message.copy(), wireFormat);
+ copy.setMessageID(newID);
+ return copy;
+ }
+
+ @Override
+ public long getMessageID() {
+ return message.getMessageId().getBrokerSequenceId();
+ }
+
+ @Override
+ public Message setMessageID(long id) {
+ message.getMessageId().setBrokerSequenceId(id);
+ return this;
+ }
+
+ @Override
+ public long getExpiration() {
+ return message.getExpiration();
+ }
+
+ @Override
+ public Message setExpiration(long expiration) {
+ message.setExpiration(expiration);
+ return this;
+ }
+
+ @Override
+ public Object getUserID() {
+ return message.getUserID();
+ }
+
+ @Override
+ public Message setUserID(Object userID) {
+ message.setUserID(userID.toString());
+ return this;
+ }
+
+ @Override
+ public boolean isDurable() {
+ return message.isPersistent();
+ }
+
+ @Override
+ public Message setDurable(boolean durable) {
+ message.setPersistent(durable);
+ return this;
+ }
+
+ @Override
+ public Persister<Message> getPersister() {
+ return OpenWireMessagePersister.INSTANCE;
+ }
+
+ @Override
+ public String getAddress() {
+ return message.getDestination().getPhysicalName();
+ }
+
+ @Override
+ public Message setAddress(String address) {
+ message.setDestination(ActiveMQDestination.createDestination(address, ActiveMQDestination.QUEUE_TYPE));
+ return this;
+ }
+
+ @Override
+ public SimpleString getAddressSimpleString() {
+ return SimpleString.toSimpleString(message.getDestination().getPhysicalName());
+ }
+
+ @Override
+ public Message setAddress(SimpleString address) {
+ message.setDestination(ActiveMQDestination.createDestination(address.toString(), ActiveMQDestination.QUEUE_TYPE));
+ return this;
+ }
+
+ @Override
+ public long getTimestamp() {
+ return message.getTimestamp();
+ }
+
+ @Override
+ public Message setTimestamp(long timestamp) {
+ message.setTimestamp(timestamp);
+ return this;
+ }
+
+ @Override
+ public byte getPriority() {
+ return message.getPriority();
+ }
+
+ @Override
+ public Message setPriority(byte priority) {
+ message.setPriority(priority);
+ return this;
+ }
+
+ @Override
+ public void receiveBuffer(ByteBuf buffer) {
+
+ }
+
+ @Override
+ public void sendBuffer(ByteBuf buffer, int deliveryCount) {
+
+ }
+
+ @Override
+ public int getPersistSize() {
+ return DataConstants.SIZE_INT + data.array().length;
+ }
+
+ @Override
+ public void persist(ActiveMQBuffer targetRecord) {
+ targetRecord.writeInt(data.writerIndex());
+ targetRecord.writeBytes(data, 0, data.writerIndex());
+ }
+
+ @Override
+ public void reloadPersistence(ActiveMQBuffer record) {
+ int size = record.readInt();
+ byte[] recordArray = new byte[size];
+ record.readBytes(recordArray);
+ this.data = Unpooled.wrappedBuffer(recordArray);
+ OpenWireFormat owf = new OpenWireFormat();
+ try {
+ message = (ActiveMQMessage) owf.unmarshal(new ChannelBufferWrapper(data));
+ } catch (IOException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Message putBooleanProperty(String key, boolean value) {
+ try {
+ message.setBooleanProperty(key, value);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putByteProperty(String key, byte value) {
+ try {
+ message.setByteProperty(key, value);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putBytesProperty(String key, byte[] value) {
+ return putObjectProperty(key, value);
+ }
+
+ @Override
+ public Message putShortProperty(String key, short value) {
+ try {
+ message.setShortProperty(key, value);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putCharProperty(String key, char value) {
+ return putStringProperty(key, Character.toString(value));
+ }
+
+ @Override
+ public Message putIntProperty(String key, int value) {
+ try {
+ message.setIntProperty(key, value);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putLongProperty(String key, long value) {
+ try {
+ message.setLongProperty(key, value);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putFloatProperty(String key, float value) {
+ try {
+ message.setFloatProperty(key, value);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putDoubleProperty(String key, double value) {
+ try {
+ message.setDoubleProperty(key, value);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putBooleanProperty(SimpleString key, boolean value) {
+ return putBooleanProperty(key.toString(), value);
+ }
+
+ @Override
+ public Message putByteProperty(SimpleString key, byte value) {
+ return putByteProperty(key.toString(), value);
+ }
+
+ @Override
+ public Message putBytesProperty(SimpleString key, byte[] value) {
+ return putBytesProperty(key.toString(), value);
+ }
+
+ @Override
+ public Message putShortProperty(SimpleString key, short value) {
+ return putShortProperty(key.toString(), value);
+ }
+
+ @Override
+ public Message putCharProperty(SimpleString key, char value) {
+ return putCharProperty(key.toString(), value);
+ }
+
+ @Override
+ public Message putIntProperty(SimpleString key, int value) {
+ return putIntProperty(key.toString(), value);
+ }
+
+ @Override
+ public Message putLongProperty(SimpleString key, long value) {
+ return putLongProperty(key.toString(), value);
+ }
+
+ @Override
+ public Message putFloatProperty(SimpleString key, float value) {
+ return putFloatProperty(key.toString(), value);
+
+ }
+
+ @Override
+ public Message putDoubleProperty(SimpleString key, double value) {
+ return putDoubleProperty(key.toString(), value);
+ }
+
+ @Override
+ public Message putStringProperty(String key, String value) {
+ try {
+ message.setStringProperty(key, value);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putObjectProperty(String key, Object value) throws ActiveMQPropertyConversionException {
+ try {
+ final Object v;
+ if (value instanceof SimpleString) {
+ v = value.toString();
+ } else {
+ v = value;
+ }
+ message.setProperty(key, value);
+ } catch (IOException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putObjectProperty(SimpleString key, Object value) throws ActiveMQPropertyConversionException {
+ return putObjectProperty(key.toString(), value);
+ }
+
+ @Override
+ public Object removeProperty(String key) {
+ try {
+ Object o = message.getProperty(key);
+ if (o != null) {
+ message.removeProperty(key);
+ }
+ return o;
+ } catch (IOException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+
+ }
+
+ @Override
+ public boolean containsProperty(String key) {
+ try {
+ return message.getProperty(key) != null;
+ } catch (IOException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Boolean getBooleanProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return message.getBooleanProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Byte getByteProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return message.getByteProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Double getDoubleProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return message.getDoubleProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Integer getIntProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return message.getIntProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Long getLongProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return message.getLongProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Object getObjectProperty(String key) {
+ try {
+ return message.getObjectProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Short getShortProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return message.getShortProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Float getFloatProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return message.getFloatProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public String getStringProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return message.getStringProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public SimpleString getSimpleStringProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return SimpleString.toSimpleString(message.getStringProperty(key));
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public byte[] getBytesProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return (byte[]) message.getObjectProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Object removeProperty(SimpleString key) {
+ return removeProperty(key.toString());
+ }
+
+ @Override
+ public boolean containsProperty(SimpleString key) {
+ return containsProperty(key.toString());
+ }
+
+ @Override
+ public Boolean getBooleanProperty(SimpleString key) throws ActiveMQPropertyConversionException {
+ return getBooleanProperty(key.toString());
+ }
+
+ @Override
+ public Byte getByteProperty(SimpleString key) throws ActiveMQPropertyConversionException {
+ return getByteProperty(key.toString());
+ }
+
+ @Override
+ public Double getDoubleProperty(SimpleString key) throws ActiveMQPropertyConversionException {
+ return getDoubleProperty(key.toString());
+ }
+
+ @Override
+ public Integer getIntProperty(SimpleString key) throws ActiveMQPropertyConversionException {
+ return getIntProperty(key.toString());
+ }
+
+ @Override
+ public Long getLongProperty(SimpleString key) throws ActiveMQPropertyConversionException {
+ return getLongProperty(key.toString());
+ }
+
+ @Override
+ public Object getObjectProperty(SimpleString key) {
+ return getObjectProperty(key.toString());
+ }
+
+ @Override
+ public Object getAnnotation(SimpleString key) {
+ return getObjectProperty(key.toString());
+ }
+
+ @Override
+ public Short getShortProperty(SimpleString key) throws ActiveMQPropertyConversionException {
+ return getShortProperty(key.toString());
+ }
+
+ @Override
+ public Float getFloatProperty(SimpleString key) throws ActiveMQPropertyConversionException {
+ return getFloatProperty(key.toString());
+ }
+
+ @Override
+ public String getStringProperty(SimpleString key) throws ActiveMQPropertyConversionException {
+ return getStringProperty(key.toString());
+ }
+
+ @Override
+ public SimpleString getSimpleStringProperty(SimpleString key) throws ActiveMQPropertyConversionException {
+ return SimpleString.toSimpleString(getStringProperty(key));
+ }
+
+ @Override
+ public byte[] getBytesProperty(SimpleString key) throws ActiveMQPropertyConversionException {
+ return getBytesProperty(key.toString());
+ }
+
+ @Override
+ public Message putStringProperty(SimpleString key, SimpleString value) {
+ return putStringProperty(key.toString(), value.toString());
+ }
+
+ @Override
+ public Message putStringProperty(SimpleString key, String value) {
+ return putStringProperty(key.toString(), value);
+ }
+
+ @Override
+ public int getEncodeSize() {
+ return 0;
+ }
+
+ @Override
+ public Set<SimpleString> getPropertyNames() {
+ try {
+ Set<SimpleString> simpleStrings = new HashSet<>();
+ Enumeration propertyNames = message.getPropertyNames();
+ while (propertyNames.hasMoreElements()) {
+ simpleStrings.add(SimpleString.toSimpleString((String) propertyNames.nextElement()));
--- End diff --
Use pool
---
[GitHub] activemq-artemis issue #1793: ARTEMIS-1498: Openwire internal headers should...
Posted by michaelandrepearce <gi...@git.apache.org>.
Github user michaelandrepearce commented on the issue:
https://github.com/apache/activemq-artemis/pull/1793
@raisaurabh great stuff, if I get a chance I’ll try have a look. If not hopefully Clebert will be able to.
---
[GitHub] activemq-artemis issue #1793: ARTEMIS-1498: Openwire internal headers should...
Posted by RaiSaurabh <gi...@git.apache.org>.
Github user RaiSaurabh commented on the issue:
https://github.com/apache/activemq-artemis/pull/1793
@michaelandrepearce I have implemented the changes asked and also checked that this persists to disk and is recoverable on restart of the broker without converting to the core message.
Could you please review it.
---
[GitHub] activemq-artemis pull request #1793: ARTEMIS-1498: Openwire internal headers...
Posted by michaelandrepearce <gi...@git.apache.org>.
Github user michaelandrepearce commented on a diff in the pull request:
https://github.com/apache/activemq-artemis/pull/1793#discussion_r165768988
--- Diff: artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireMessage.java ---
@@ -0,0 +1,1152 @@
+/**
+ * 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.activemq.artemis.core.protocol.openwire;
+import java.io.InputStream;
+import java.nio.ByteBuffer;
+import java.util.LinkedList;
+import java.util.Set;
+
+import io.netty.buffer.ByteBuf;
+import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
+import org.apache.activemq.artemis.api.core.ActiveMQBuffers;
+import org.apache.activemq.artemis.api.core.ActiveMQException;
+import org.apache.activemq.artemis.api.core.ActiveMQPropertyConversionException;
+import org.apache.activemq.artemis.api.core.ICoreMessage;
+import org.apache.activemq.artemis.api.core.Message;
+import org.apache.activemq.artemis.api.core.RefCountMessage;
+import org.apache.activemq.artemis.api.core.RoutingType;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.buffers.impl.ChannelBufferWrapper;
+import org.apache.activemq.artemis.core.buffers.impl.ResetLimitWrappedActiveMQBuffer;
+import org.apache.activemq.artemis.core.message.LargeBodyEncoder;
+import org.apache.activemq.artemis.core.message.impl.CoreMessageObjectPools;
+import org.apache.activemq.artemis.core.persistence.Persister;
+import org.apache.activemq.artemis.core.protocol.core.impl.PacketImpl;
+import org.apache.activemq.artemis.reader.MessageUtil;
+import org.apache.activemq.artemis.utils.DataConstants;
+import org.apache.activemq.artemis.utils.UUID;
+import org.apache.activemq.artemis.utils.collections.TypedProperties;
+import org.jboss.logging.Logger;
+
+/** Note: you shouldn't change properties using multi-threads. Change your properties before you can send it to multiple
+ * consumers */
+public class OpenWireMessage extends RefCountMessage {
+ public static final int BUFFER_HEADER_SPACE = PacketImpl.PACKET_HEADERS_SIZE;
+
+ private volatile int memoryEstimate = -1;
+ private static final Logger logger = Logger.getLogger(OpenWireMessage.class);
+
+ // There's an integer with the number of bytes for the body
+ public static final int BODY_OFFSET = DataConstants.SIZE_INT;
+
+ /** That is the encode for the whole message, including properties..
+ it does not include the buffer for the Packet send and receive header on core protocol */
+ protected ByteBuf buffer;
+
+ private volatile boolean validBuffer = false;
--- End diff --
Lets merge it, and then sort it later IMO. Im keen to get this big PR in, as long as functional and no-regressions im happy personally.
---
[GitHub] activemq-artemis pull request #1793: ARTEMIS-1498: Openwire internal headers...
Posted by franz1981 <gi...@git.apache.org>.
Github user franz1981 commented on a diff in the pull request:
https://github.com/apache/activemq-artemis/pull/1793#discussion_r165600483
--- Diff: artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireCoreConverter.java ---
@@ -0,0 +1,572 @@
+/*
+ * 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.activemq.artemis.core.protocol.openwire;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Arrays;
+import java.util.Map;
+import java.util.Set;
+import java.util.Map.Entry;
+import java.util.zip.Deflater;
+import java.util.zip.DeflaterOutputStream;
+import java.util.zip.Inflater;
+import java.util.zip.InflaterInputStream;
+import java.util.zip.InflaterOutputStream;
+
+import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
+import org.apache.activemq.artemis.api.core.FilterConstants;
+import org.apache.activemq.artemis.api.core.ICoreMessage;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.message.impl.CoreMessage;
+import org.apache.activemq.artemis.core.message.impl.CoreMessageObjectPools;
+import org.apache.activemq.artemis.reader.MessageUtil;
+import org.apache.activemq.artemis.utils.DataConstants;
+import org.apache.activemq.artemis.utils.collections.TypedProperties;
+import org.apache.activemq.command.CommandTypes;
+import org.apache.activemq.util.ByteArrayInputStream;
+import org.apache.activemq.util.ByteSequence;
+import org.apache.activemq.util.ByteSequenceData;
+import org.apache.activemq.util.MarshallingSupport;
+import org.fusesource.hawtbuf.UTF8Buffer;
+
+public class OpenWireCoreConverter {
+ public static final SimpleString JMS_GROUPSEQ = new SimpleString("JMSXGroupSequence");
+ public static final SimpleString JMS_XGROUPSEQ = new SimpleString("JMSXGroupSeq");
+ public static final SimpleString JMS_GROUPID = new SimpleString("JMSXGroupID");
+ private static final SimpleString AMQP_REPLYTOGROUPID = new SimpleString("JMS_AMQP_ReplyToGroupID");
+ public static final SimpleString JMS_CONTENT_TYPE = new SimpleString("JMS_AMQP_ContentType");
+ private static final SimpleString AMQP_CONTETTYPE = new SimpleString("JMS_AMQP_CONTENT_TYPE");
+
+ public static ICoreMessage toCore(final OpenWireMessage openwireMessage, final CoreMessageObjectPools coreMessageObjectPools)
+ throws Exception {
+ CoreMessage coreMessage = new CoreMessage(-1, openwireMessage.getMessageSize(), coreMessageObjectPools);
+ final SimpleString type = openwireMessage.getSimpleStringProperty(OpenWireMessageConverter.JMS_TYPE_PROPERTY);
+ if (type != null) {
+ coreMessage.putStringProperty(OpenWireMessageConverter.JMS_TYPE_PROPERTY, type);
+ }
+ coreMessage.setDurable(openwireMessage.isDurable());
+ coreMessage.setExpiration(openwireMessage.getExpiration());
+ coreMessage.setPriority(openwireMessage.getPriority());
+ coreMessage.setTimestamp(openwireMessage.getTimestamp());
+ final ActiveMQBuffer openWireMessageBuffer = openwireMessage.getReadOnlyBodyBuffer();
+ final boolean isCompressed = openwireMessage.getBooleanProperty(OpenWireMessageConverter.AMQ_MSG_COMPRESSED);
+ final byte coreType = openwireMessage.getType();
+ coreMessage.setType(coreType);
+ openWireMessageBuffer.resetReaderIndex();
+ final byte[] bytes;
+ bytes = readContentFromBody(openWireMessageBuffer, coreType, isCompressed);
+ if (bytes != null) {
+ ByteSequence content = new ByteSequence(bytes);
+ ActiveMQBuffer coreMessageBuffer = coreMessage.getBodyBuffer();
+ writeContentIntoBody(coreMessageBuffer, content, coreType, isCompressed);
+ }
+ coreMessage.putObjectProperty(FilterConstants.NATIVE_MESSAGE_ID,
+ openwireMessage.getObjectProperty(FilterConstants.NATIVE_MESSAGE_ID));
+ coreMessage.setMessageID(openwireMessage.getMessageID());
+ final String corrId = openwireMessage.getStringProperty(OpenWireMessageConverter.JMS_CORRELATION_ID_PROPERTY);
+ if (corrId != null) {
+ coreMessage.putStringProperty(OpenWireMessageConverter.JMS_CORRELATION_ID_PROPERTY, corrId);
+ }
+ final String groupId = openwireMessage.getStringProperty(org.apache.activemq.artemis.api.core.Message.HDR_GROUP_ID);
+ if (groupId != null) {
+ coreMessage.putStringProperty(org.apache.activemq.artemis.api.core.Message.HDR_GROUP_ID,
+ new SimpleString(groupId));
+ coreMessage.putStringProperty(JMS_GROUPID, new SimpleString(groupId));
--- End diff --
Same as the comment above
---
[GitHub] activemq-artemis issue #1793: ARTEMIS-1498: Openwire internal headers should...
Posted by michaelandrepearce <gi...@git.apache.org>.
Github user michaelandrepearce commented on the issue:
https://github.com/apache/activemq-artemis/pull/1793
@franz1981 im happy to do the merge, but i would like a thumbs up from you before i do.
---
[GitHub] activemq-artemis pull request #1793: ARTEMIS-1498: Openwire internal headers...
Posted by michaelandrepearce <gi...@git.apache.org>.
Github user michaelandrepearce commented on a diff in the pull request:
https://github.com/apache/activemq-artemis/pull/1793#discussion_r165768314
--- Diff: artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireCoreConverter.java ---
@@ -0,0 +1,572 @@
+/*
+ * 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.activemq.artemis.core.protocol.openwire;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Arrays;
+import java.util.Map;
+import java.util.Set;
+import java.util.Map.Entry;
+import java.util.zip.Deflater;
+import java.util.zip.DeflaterOutputStream;
+import java.util.zip.Inflater;
+import java.util.zip.InflaterInputStream;
+import java.util.zip.InflaterOutputStream;
+
+import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
+import org.apache.activemq.artemis.api.core.FilterConstants;
+import org.apache.activemq.artemis.api.core.ICoreMessage;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.message.impl.CoreMessage;
+import org.apache.activemq.artemis.core.message.impl.CoreMessageObjectPools;
+import org.apache.activemq.artemis.reader.MessageUtil;
+import org.apache.activemq.artemis.utils.DataConstants;
+import org.apache.activemq.artemis.utils.collections.TypedProperties;
+import org.apache.activemq.command.CommandTypes;
+import org.apache.activemq.util.ByteArrayInputStream;
+import org.apache.activemq.util.ByteSequence;
+import org.apache.activemq.util.ByteSequenceData;
+import org.apache.activemq.util.MarshallingSupport;
+import org.fusesource.hawtbuf.UTF8Buffer;
+
+public class OpenWireCoreConverter {
+ public static final SimpleString JMS_GROUPSEQ = new SimpleString("JMSXGroupSequence");
+ public static final SimpleString JMS_XGROUPSEQ = new SimpleString("JMSXGroupSeq");
+ public static final SimpleString JMS_GROUPID = new SimpleString("JMSXGroupID");
+ private static final SimpleString AMQP_REPLYTOGROUPID = new SimpleString("JMS_AMQP_ReplyToGroupID");
+ public static final SimpleString JMS_CONTENT_TYPE = new SimpleString("JMS_AMQP_ContentType");
+ private static final SimpleString AMQP_CONTETTYPE = new SimpleString("JMS_AMQP_CONTENT_TYPE");
+
+ public static ICoreMessage toCore(final OpenWireMessage openwireMessage, final CoreMessageObjectPools coreMessageObjectPools)
+ throws Exception {
+ CoreMessage coreMessage = new CoreMessage(-1, openwireMessage.getMessageSize(), coreMessageObjectPools);
+ final SimpleString type = openwireMessage.getSimpleStringProperty(OpenWireMessageConverter.JMS_TYPE_PROPERTY);
+ if (type != null) {
+ coreMessage.putStringProperty(OpenWireMessageConverter.JMS_TYPE_PROPERTY, type);
+ }
+ coreMessage.setDurable(openwireMessage.isDurable());
+ coreMessage.setExpiration(openwireMessage.getExpiration());
+ coreMessage.setPriority(openwireMessage.getPriority());
+ coreMessage.setTimestamp(openwireMessage.getTimestamp());
+ final ActiveMQBuffer openWireMessageBuffer = openwireMessage.getReadOnlyBodyBuffer();
+ final boolean isCompressed = openwireMessage.getBooleanProperty(OpenWireMessageConverter.AMQ_MSG_COMPRESSED);
+ final byte coreType = openwireMessage.getType();
+ coreMessage.setType(coreType);
+ openWireMessageBuffer.resetReaderIndex();
+ final byte[] bytes;
+ bytes = readContentFromBody(openWireMessageBuffer, coreType, isCompressed);
+ if (bytes != null) {
+ ByteSequence content = new ByteSequence(bytes);
+ ActiveMQBuffer coreMessageBuffer = coreMessage.getBodyBuffer();
+ writeContentIntoBody(coreMessageBuffer, content, coreType, isCompressed);
+ }
+ coreMessage.putObjectProperty(FilterConstants.NATIVE_MESSAGE_ID,
+ openwireMessage.getObjectProperty(FilterConstants.NATIVE_MESSAGE_ID));
+ coreMessage.setMessageID(openwireMessage.getMessageID());
+ final String corrId = openwireMessage.getStringProperty(OpenWireMessageConverter.JMS_CORRELATION_ID_PROPERTY);
+ if (corrId != null) {
+ coreMessage.putStringProperty(OpenWireMessageConverter.JMS_CORRELATION_ID_PROPERTY, corrId);
+ }
+ final String groupId = openwireMessage.getStringProperty(org.apache.activemq.artemis.api.core.Message.HDR_GROUP_ID);
+ if (groupId != null) {
+ coreMessage.putStringProperty(org.apache.activemq.artemis.api.core.Message.HDR_GROUP_ID,
+ new SimpleString(groupId));
--- End diff --
+1 good suggestion, i think maybe we can do this optimisation later though, this is a large change, and improves things anyhow. e.g. why dont we get this PR in, and then break loose on perf later?
---
[GitHub] activemq-artemis pull request #1793: ARTEMIS-1498: Openwire internal headers...
Posted by franz1981 <gi...@git.apache.org>.
Github user franz1981 commented on a diff in the pull request:
https://github.com/apache/activemq-artemis/pull/1793#discussion_r165602075
--- Diff: artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireCoreConverter.java ---
@@ -0,0 +1,572 @@
+/*
+ * 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.activemq.artemis.core.protocol.openwire;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Arrays;
+import java.util.Map;
+import java.util.Set;
+import java.util.Map.Entry;
+import java.util.zip.Deflater;
+import java.util.zip.DeflaterOutputStream;
+import java.util.zip.Inflater;
+import java.util.zip.InflaterInputStream;
+import java.util.zip.InflaterOutputStream;
+
+import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
+import org.apache.activemq.artemis.api.core.FilterConstants;
+import org.apache.activemq.artemis.api.core.ICoreMessage;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.message.impl.CoreMessage;
+import org.apache.activemq.artemis.core.message.impl.CoreMessageObjectPools;
+import org.apache.activemq.artemis.reader.MessageUtil;
+import org.apache.activemq.artemis.utils.DataConstants;
+import org.apache.activemq.artemis.utils.collections.TypedProperties;
+import org.apache.activemq.command.CommandTypes;
+import org.apache.activemq.util.ByteArrayInputStream;
+import org.apache.activemq.util.ByteSequence;
+import org.apache.activemq.util.ByteSequenceData;
+import org.apache.activemq.util.MarshallingSupport;
+import org.fusesource.hawtbuf.UTF8Buffer;
+
+public class OpenWireCoreConverter {
+ public static final SimpleString JMS_GROUPSEQ = new SimpleString("JMSXGroupSequence");
+ public static final SimpleString JMS_XGROUPSEQ = new SimpleString("JMSXGroupSeq");
+ public static final SimpleString JMS_GROUPID = new SimpleString("JMSXGroupID");
+ private static final SimpleString AMQP_REPLYTOGROUPID = new SimpleString("JMS_AMQP_ReplyToGroupID");
+ public static final SimpleString JMS_CONTENT_TYPE = new SimpleString("JMS_AMQP_ContentType");
+ private static final SimpleString AMQP_CONTETTYPE = new SimpleString("JMS_AMQP_CONTENT_TYPE");
+
+ public static ICoreMessage toCore(final OpenWireMessage openwireMessage, final CoreMessageObjectPools coreMessageObjectPools)
+ throws Exception {
+ CoreMessage coreMessage = new CoreMessage(-1, openwireMessage.getMessageSize(), coreMessageObjectPools);
+ final SimpleString type = openwireMessage.getSimpleStringProperty(OpenWireMessageConverter.JMS_TYPE_PROPERTY);
+ if (type != null) {
+ coreMessage.putStringProperty(OpenWireMessageConverter.JMS_TYPE_PROPERTY, type);
+ }
+ coreMessage.setDurable(openwireMessage.isDurable());
+ coreMessage.setExpiration(openwireMessage.getExpiration());
+ coreMessage.setPriority(openwireMessage.getPriority());
+ coreMessage.setTimestamp(openwireMessage.getTimestamp());
+ final ActiveMQBuffer openWireMessageBuffer = openwireMessage.getReadOnlyBodyBuffer();
+ final boolean isCompressed = openwireMessage.getBooleanProperty(OpenWireMessageConverter.AMQ_MSG_COMPRESSED);
+ final byte coreType = openwireMessage.getType();
+ coreMessage.setType(coreType);
+ openWireMessageBuffer.resetReaderIndex();
+ final byte[] bytes;
+ bytes = readContentFromBody(openWireMessageBuffer, coreType, isCompressed);
+ if (bytes != null) {
+ ByteSequence content = new ByteSequence(bytes);
+ ActiveMQBuffer coreMessageBuffer = coreMessage.getBodyBuffer();
+ writeContentIntoBody(coreMessageBuffer, content, coreType, isCompressed);
+ }
+ coreMessage.putObjectProperty(FilterConstants.NATIVE_MESSAGE_ID,
+ openwireMessage.getObjectProperty(FilterConstants.NATIVE_MESSAGE_ID));
+ coreMessage.setMessageID(openwireMessage.getMessageID());
+ final String corrId = openwireMessage.getStringProperty(OpenWireMessageConverter.JMS_CORRELATION_ID_PROPERTY);
+ if (corrId != null) {
+ coreMessage.putStringProperty(OpenWireMessageConverter.JMS_CORRELATION_ID_PROPERTY, corrId);
+ }
+ final String groupId = openwireMessage.getStringProperty(org.apache.activemq.artemis.api.core.Message.HDR_GROUP_ID);
+ if (groupId != null) {
+ coreMessage.putStringProperty(org.apache.activemq.artemis.api.core.Message.HDR_GROUP_ID,
+ new SimpleString(groupId));
+ coreMessage.putStringProperty(JMS_GROUPID, new SimpleString(groupId));
+ }
+ if (openwireMessage.containsProperty(JMS_GROUPSEQ)) {
+ coreMessage.putIntProperty(JMS_GROUPSEQ, openwireMessage.getIntProperty(JMS_GROUPSEQ));
+ coreMessage.putIntProperty(JMS_XGROUPSEQ, openwireMessage.getIntProperty(JMS_GROUPSEQ));
+ }
+ final SimpleString lastValueProperty = openwireMessage.getLastValueProperty();
+ if (lastValueProperty != null) {
+ coreMessage.setLastValueProperty(lastValueProperty);
+ }
+ coreMessage.putObjectProperty(AMQP_CONTETTYPE, openwireMessage.getObjectProperty(JMS_CONTENT_TYPE));
+ coreMessage.putObjectProperty(AMQP_REPLYTOGROUPID, openwireMessage.getObjectProperty(AMQP_REPLYTOGROUPID));
+ coreMessage.setReplyTo(openwireMessage.getReplyTo());
+ coreMessage.setUserID(openwireMessage.getUserID());
+ coreMessage.setDurable(openwireMessage.isDurable());
+ coreMessage.setAddress(openwireMessage.getAddress());
+ coreMessage.putObjectProperty(org.apache.activemq.artemis.api.core.Message.HDR_GROUP_ID,
+ (SimpleString) openwireMessage
+ .getObjectProperty(org.apache.activemq.artemis.api.core.Message.HDR_GROUP_ID));
+ copyAllProperties(openwireMessage, coreMessage);
+ return coreMessage;
+ }
+
+ public static byte toCoreType(byte amqType) {
+ switch (amqType) {
+ case CommandTypes.ACTIVEMQ_BLOB_MESSAGE:
+ throw new IllegalStateException("We don't support BLOB type yet!");
+ case CommandTypes.ACTIVEMQ_BYTES_MESSAGE:
+ return org.apache.activemq.artemis.api.core.Message.BYTES_TYPE;
+ case CommandTypes.ACTIVEMQ_MAP_MESSAGE:
+ return org.apache.activemq.artemis.api.core.Message.MAP_TYPE;
+ case CommandTypes.ACTIVEMQ_OBJECT_MESSAGE:
+ return org.apache.activemq.artemis.api.core.Message.OBJECT_TYPE;
+ case CommandTypes.ACTIVEMQ_STREAM_MESSAGE:
+ return org.apache.activemq.artemis.api.core.Message.STREAM_TYPE;
+ case CommandTypes.ACTIVEMQ_TEXT_MESSAGE:
+ return org.apache.activemq.artemis.api.core.Message.TEXT_TYPE;
+ case CommandTypes.ACTIVEMQ_MESSAGE:
+ return org.apache.activemq.artemis.api.core.Message.DEFAULT_TYPE;
+ default:
+ throw new IllegalStateException("Unknown ActiveMQ Artemis message type: " + amqType);
+ }
+ }
+
+ public static void loadMapIntoProperties(TypedProperties props, Map<String, Object> map) {
+ for (Entry<String, Object> entry : map.entrySet()) {
+ SimpleString key = new SimpleString(entry.getKey());
+ Object value = entry.getValue();
+ if (value instanceof UTF8Buffer) {
+ value = ((UTF8Buffer) value).toString();
--- End diff --
This cast is redundant: it was in the original code too, but it could dropped now
---
[GitHub] activemq-artemis pull request #1793: ARTEMIS-1498: Openwire internal headers...
Posted by franz1981 <gi...@git.apache.org>.
Github user franz1981 commented on a diff in the pull request:
https://github.com/apache/activemq-artemis/pull/1793#discussion_r165600076
--- Diff: artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireCoreConverter.java ---
@@ -0,0 +1,572 @@
+/*
+ * 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.activemq.artemis.core.protocol.openwire;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataInputStream;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.util.Arrays;
+import java.util.Map;
+import java.util.Set;
+import java.util.Map.Entry;
+import java.util.zip.Deflater;
+import java.util.zip.DeflaterOutputStream;
+import java.util.zip.Inflater;
+import java.util.zip.InflaterInputStream;
+import java.util.zip.InflaterOutputStream;
+
+import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
+import org.apache.activemq.artemis.api.core.FilterConstants;
+import org.apache.activemq.artemis.api.core.ICoreMessage;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.message.impl.CoreMessage;
+import org.apache.activemq.artemis.core.message.impl.CoreMessageObjectPools;
+import org.apache.activemq.artemis.reader.MessageUtil;
+import org.apache.activemq.artemis.utils.DataConstants;
+import org.apache.activemq.artemis.utils.collections.TypedProperties;
+import org.apache.activemq.command.CommandTypes;
+import org.apache.activemq.util.ByteArrayInputStream;
+import org.apache.activemq.util.ByteSequence;
+import org.apache.activemq.util.ByteSequenceData;
+import org.apache.activemq.util.MarshallingSupport;
+import org.fusesource.hawtbuf.UTF8Buffer;
+
+public class OpenWireCoreConverter {
+ public static final SimpleString JMS_GROUPSEQ = new SimpleString("JMSXGroupSequence");
+ public static final SimpleString JMS_XGROUPSEQ = new SimpleString("JMSXGroupSeq");
+ public static final SimpleString JMS_GROUPID = new SimpleString("JMSXGroupID");
+ private static final SimpleString AMQP_REPLYTOGROUPID = new SimpleString("JMS_AMQP_ReplyToGroupID");
+ public static final SimpleString JMS_CONTENT_TYPE = new SimpleString("JMS_AMQP_ContentType");
+ private static final SimpleString AMQP_CONTETTYPE = new SimpleString("JMS_AMQP_CONTENT_TYPE");
+
+ public static ICoreMessage toCore(final OpenWireMessage openwireMessage, final CoreMessageObjectPools coreMessageObjectPools)
+ throws Exception {
+ CoreMessage coreMessage = new CoreMessage(-1, openwireMessage.getMessageSize(), coreMessageObjectPools);
+ final SimpleString type = openwireMessage.getSimpleStringProperty(OpenWireMessageConverter.JMS_TYPE_PROPERTY);
+ if (type != null) {
+ coreMessage.putStringProperty(OpenWireMessageConverter.JMS_TYPE_PROPERTY, type);
+ }
+ coreMessage.setDurable(openwireMessage.isDurable());
+ coreMessage.setExpiration(openwireMessage.getExpiration());
+ coreMessage.setPriority(openwireMessage.getPriority());
+ coreMessage.setTimestamp(openwireMessage.getTimestamp());
+ final ActiveMQBuffer openWireMessageBuffer = openwireMessage.getReadOnlyBodyBuffer();
+ final boolean isCompressed = openwireMessage.getBooleanProperty(OpenWireMessageConverter.AMQ_MSG_COMPRESSED);
+ final byte coreType = openwireMessage.getType();
+ coreMessage.setType(coreType);
+ openWireMessageBuffer.resetReaderIndex();
+ final byte[] bytes;
+ bytes = readContentFromBody(openWireMessageBuffer, coreType, isCompressed);
+ if (bytes != null) {
+ ByteSequence content = new ByteSequence(bytes);
+ ActiveMQBuffer coreMessageBuffer = coreMessage.getBodyBuffer();
+ writeContentIntoBody(coreMessageBuffer, content, coreType, isCompressed);
+ }
+ coreMessage.putObjectProperty(FilterConstants.NATIVE_MESSAGE_ID,
+ openwireMessage.getObjectProperty(FilterConstants.NATIVE_MESSAGE_ID));
+ coreMessage.setMessageID(openwireMessage.getMessageID());
+ final String corrId = openwireMessage.getStringProperty(OpenWireMessageConverter.JMS_CORRELATION_ID_PROPERTY);
+ if (corrId != null) {
+ coreMessage.putStringProperty(OpenWireMessageConverter.JMS_CORRELATION_ID_PROPERTY, corrId);
+ }
+ final String groupId = openwireMessage.getStringProperty(org.apache.activemq.artemis.api.core.Message.HDR_GROUP_ID);
+ if (groupId != null) {
+ coreMessage.putStringProperty(org.apache.activemq.artemis.api.core.Message.HDR_GROUP_ID,
+ new SimpleString(groupId));
--- End diff --
`new SimpleString(groupId)` could be replaced by `coreMessageObjectPools.getGroupIdStringSimpleStringPool().getOrCreate(groupId)`
---
[GitHub] activemq-artemis issue #1793: ARTEMIS-1498: Openwire internal headers should...
Posted by michaelandrepearce <gi...@git.apache.org>.
Github user michaelandrepearce commented on the issue:
https://github.com/apache/activemq-artemis/pull/1793
@franz1981 I think there’s a general Jenkins PR build issue. If you have direct ping ability or nudge directly Martyn or Clebert or someone on PM ?
---
[GitHub] activemq-artemis pull request #1793: ARTEMIS-1498: Openwire internal headers...
Posted by RaiSaurabh <gi...@git.apache.org>.
Github user RaiSaurabh commented on a diff in the pull request:
https://github.com/apache/activemq-artemis/pull/1793#discussion_r168675045
--- Diff: artemis-protocols/artemis-openwire-protocol/src/main/java/org/apache/activemq/artemis/core/protocol/openwire/OpenWireMessage.java ---
@@ -0,0 +1,673 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.activemq.artemis.core.protocol.openwire;
+
+import javax.jms.JMSException;
+import java.io.IOException;
+import java.util.Enumeration;
+import java.util.HashSet;
+import java.util.Set;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
+import org.apache.activemq.artemis.api.core.ActiveMQException;
+import org.apache.activemq.artemis.api.core.ActiveMQPropertyConversionException;
+import org.apache.activemq.artemis.api.core.ICoreMessage;
+import org.apache.activemq.artemis.api.core.Message;
+import org.apache.activemq.artemis.api.core.RefCountMessage;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.buffers.impl.ChannelBufferWrapper;
+import org.apache.activemq.artemis.core.message.impl.CoreMessageObjectPools;
+import org.apache.activemq.artemis.core.persistence.Persister;
+import org.apache.activemq.artemis.utils.DataConstants;
+import org.apache.activemq.command.ActiveMQDestination;
+import org.apache.activemq.command.ActiveMQMessage;
+import org.apache.activemq.openwire.OpenWireFormat;
+import org.apache.activemq.util.ByteSequence;
+import org.apache.activemq.wireformat.WireFormat;
+
+public class OpenWireMessage extends RefCountMessage {
+
+ private org.apache.activemq.command.ActiveMQMessage message;
+ private WireFormat wireFormat;
+ ByteBuf data;
+ boolean bufferValid;
+
+
+ public OpenWireMessage(org.apache.activemq.command.ActiveMQMessage message, WireFormat wireFormat) {
+ this.message = message;
+ this.wireFormat = wireFormat;
+ try {
+ ByteSequence byteSequence = this.wireFormat.marshal(message);
+ setBuffer(Unpooled.copiedBuffer(byteSequence.getData(), byteSequence.getOffset(), byteSequence.getLength()));
+ } catch (IOException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ public OpenWireMessage() {
+ }
+
+ public void setMarsheller(WireFormat wireFormat) {
+ this.wireFormat = wireFormat;
+ }
+
+ public ActiveMQMessage getAMQMessage() {
+ if (message == null) {
+ if (data != null) {
+ try {
+ message = (ActiveMQMessage) wireFormat.unmarshal(new ChannelBufferWrapper(data));
+ } catch (IOException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return message;
+ } else {
+ return new ActiveMQMessage();
+ }
+ }
+ return message;
+ }
+
+ @Override
+ public void messageChanged() {
+ }
+
+ @Override
+ public Long getScheduledDeliveryTime() {
+ return message.getArrival();
+ }
+
+ @Override
+ public SimpleString getReplyTo() {
+ return SimpleString.toSimpleString(message.getReplyTo().getPhysicalName());
+ }
+
+ @Override
+ public Message setReplyTo(SimpleString address) {
+ message.setReplyTo(ActiveMQDestination.createDestination(address.toString(), ActiveMQDestination.QUEUE_TYPE));
+ return this;
+ }
+
+ @Override
+ public Message setBuffer(ByteBuf buffer) {
+ this.data = buffer;
+ return this;
+ }
+
+ @Override
+ public ByteBuf getBuffer() {
+ if (data == null) {
+ return null;
+ } else {
+ return Unpooled.wrappedBuffer(data);
+ }
+ }
+
+ @Override
+ public OpenWireMessage copy() {
+ return new OpenWireMessage((ActiveMQMessage)message.copy(), wireFormat);
+ }
+
+ @Override
+ public Message copy(long newID) {
+ OpenWireMessage copy = new OpenWireMessage((ActiveMQMessage)message.copy(), wireFormat);
+ copy.setMessageID(newID);
+ return copy;
+ }
+
+ @Override
+ public long getMessageID() {
+ return message.getMessageId().getBrokerSequenceId();
+ }
+
+ @Override
+ public Message setMessageID(long id) {
+ message.getMessageId().setBrokerSequenceId(id);
+ return this;
+ }
+
+ @Override
+ public long getExpiration() {
+ return message.getExpiration();
+ }
+
+ @Override
+ public Message setExpiration(long expiration) {
+ message.setExpiration(expiration);
+ return this;
+ }
+
+ @Override
+ public Object getUserID() {
+ return message.getUserID();
+ }
+
+ @Override
+ public Message setUserID(Object userID) {
+ message.setUserID(userID.toString());
+ return this;
+ }
+
+ @Override
+ public boolean isDurable() {
+ return message.isPersistent();
+ }
+
+ @Override
+ public Message setDurable(boolean durable) {
+ message.setPersistent(durable);
+ return this;
+ }
+
+ @Override
+ public Persister<Message> getPersister() {
+ return OpenWireMessagePersister.INSTANCE;
+ }
+
+ @Override
+ public String getAddress() {
+ return message.getDestination().getPhysicalName();
+ }
+
+ @Override
+ public Message setAddress(String address) {
+ message.setDestination(ActiveMQDestination.createDestination(address, ActiveMQDestination.QUEUE_TYPE));
+ return this;
+ }
+
+ @Override
+ public SimpleString getAddressSimpleString() {
+ return SimpleString.toSimpleString(message.getDestination().getPhysicalName());
+ }
+
+ @Override
+ public Message setAddress(SimpleString address) {
+ message.setDestination(ActiveMQDestination.createDestination(address.toString(), ActiveMQDestination.QUEUE_TYPE));
+ return this;
+ }
+
+ @Override
+ public long getTimestamp() {
+ return message.getTimestamp();
+ }
+
+ @Override
+ public Message setTimestamp(long timestamp) {
+ message.setTimestamp(timestamp);
+ return this;
+ }
+
+ @Override
+ public byte getPriority() {
+ return message.getPriority();
+ }
+
+ @Override
+ public Message setPriority(byte priority) {
+ message.setPriority(priority);
+ return this;
+ }
+
+ @Override
+ public void receiveBuffer(ByteBuf buffer) {
+
+ }
+
+ @Override
+ public void sendBuffer(ByteBuf buffer, int deliveryCount) {
+
+ }
+
+ @Override
+ public int getPersistSize() {
+ return DataConstants.SIZE_INT + data.array().length;
+ }
+
+ @Override
+ public void persist(ActiveMQBuffer targetRecord) {
+ targetRecord.writeInt(data.writerIndex());
+ targetRecord.writeBytes(data, 0, data.writerIndex());
+ }
+
+ @Override
+ public void reloadPersistence(ActiveMQBuffer record) {
+ int size = record.readInt();
+ byte[] recordArray = new byte[size];
+ record.readBytes(recordArray);
+ this.data = Unpooled.wrappedBuffer(recordArray);
+ OpenWireFormat owf = new OpenWireFormat();
+ try {
+ message = (ActiveMQMessage) owf.unmarshal(new ChannelBufferWrapper(data));
+ } catch (IOException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Message putBooleanProperty(String key, boolean value) {
+ try {
+ message.setBooleanProperty(key, value);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putByteProperty(String key, byte value) {
+ try {
+ message.setByteProperty(key, value);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putBytesProperty(String key, byte[] value) {
+ return putObjectProperty(key, value);
+ }
+
+ @Override
+ public Message putShortProperty(String key, short value) {
+ try {
+ message.setShortProperty(key, value);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putCharProperty(String key, char value) {
+ return putStringProperty(key, Character.toString(value));
+ }
+
+ @Override
+ public Message putIntProperty(String key, int value) {
+ try {
+ message.setIntProperty(key, value);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putLongProperty(String key, long value) {
+ try {
+ message.setLongProperty(key, value);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putFloatProperty(String key, float value) {
+ try {
+ message.setFloatProperty(key, value);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putDoubleProperty(String key, double value) {
+ try {
+ message.setDoubleProperty(key, value);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putBooleanProperty(SimpleString key, boolean value) {
+ return putBooleanProperty(key.toString(), value);
+ }
+
+ @Override
+ public Message putByteProperty(SimpleString key, byte value) {
+ return putByteProperty(key.toString(), value);
+ }
+
+ @Override
+ public Message putBytesProperty(SimpleString key, byte[] value) {
+ return putBytesProperty(key.toString(), value);
+ }
+
+ @Override
+ public Message putShortProperty(SimpleString key, short value) {
+ return putShortProperty(key.toString(), value);
+ }
+
+ @Override
+ public Message putCharProperty(SimpleString key, char value) {
+ return putCharProperty(key.toString(), value);
+ }
+
+ @Override
+ public Message putIntProperty(SimpleString key, int value) {
+ return putIntProperty(key.toString(), value);
+ }
+
+ @Override
+ public Message putLongProperty(SimpleString key, long value) {
+ return putLongProperty(key.toString(), value);
+ }
+
+ @Override
+ public Message putFloatProperty(SimpleString key, float value) {
+ return putFloatProperty(key.toString(), value);
+
+ }
+
+ @Override
+ public Message putDoubleProperty(SimpleString key, double value) {
+ return putDoubleProperty(key.toString(), value);
+ }
+
+ @Override
+ public Message putStringProperty(String key, String value) {
+ try {
+ message.setStringProperty(key, value);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putObjectProperty(String key, Object value) throws ActiveMQPropertyConversionException {
+ try {
+ final Object v;
+ if (value instanceof SimpleString) {
+ v = value.toString();
+ } else {
+ v = value;
+ }
+ message.setProperty(key, value);
+ } catch (IOException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ return this;
+ }
+
+ @Override
+ public Message putObjectProperty(SimpleString key, Object value) throws ActiveMQPropertyConversionException {
+ return putObjectProperty(key.toString(), value);
+ }
+
+ @Override
+ public Object removeProperty(String key) {
+ try {
+ Object o = message.getProperty(key);
+ if (o != null) {
+ message.removeProperty(key);
+ }
+ return o;
+ } catch (IOException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+
+ }
+
+ @Override
+ public boolean containsProperty(String key) {
+ try {
+ return message.getProperty(key) != null;
+ } catch (IOException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Boolean getBooleanProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return message.getBooleanProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Byte getByteProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return message.getByteProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Double getDoubleProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return message.getDoubleProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Integer getIntProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return message.getIntProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Long getLongProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return message.getLongProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Object getObjectProperty(String key) {
+ try {
+ return message.getObjectProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Short getShortProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return message.getShortProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Float getFloatProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return message.getFloatProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public String getStringProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return message.getStringProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public SimpleString getSimpleStringProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return SimpleString.toSimpleString(message.getStringProperty(key));
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public byte[] getBytesProperty(String key) throws ActiveMQPropertyConversionException {
+ try {
+ return (byte[]) message.getObjectProperty(key);
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public Object removeProperty(SimpleString key) {
+ return removeProperty(key.toString());
+ }
+
+ @Override
+ public boolean containsProperty(SimpleString key) {
+ return containsProperty(key.toString());
+ }
+
+ @Override
+ public Boolean getBooleanProperty(SimpleString key) throws ActiveMQPropertyConversionException {
+ return getBooleanProperty(key.toString());
+ }
+
+ @Override
+ public Byte getByteProperty(SimpleString key) throws ActiveMQPropertyConversionException {
+ return getByteProperty(key.toString());
+ }
+
+ @Override
+ public Double getDoubleProperty(SimpleString key) throws ActiveMQPropertyConversionException {
+ return getDoubleProperty(key.toString());
+ }
+
+ @Override
+ public Integer getIntProperty(SimpleString key) throws ActiveMQPropertyConversionException {
+ return getIntProperty(key.toString());
+ }
+
+ @Override
+ public Long getLongProperty(SimpleString key) throws ActiveMQPropertyConversionException {
+ return getLongProperty(key.toString());
+ }
+
+ @Override
+ public Object getObjectProperty(SimpleString key) {
+ return getObjectProperty(key.toString());
+ }
+
+ @Override
+ public Object getAnnotation(SimpleString key) {
+ return getObjectProperty(key.toString());
+ }
+
+ @Override
+ public Short getShortProperty(SimpleString key) throws ActiveMQPropertyConversionException {
+ return getShortProperty(key.toString());
+ }
+
+ @Override
+ public Float getFloatProperty(SimpleString key) throws ActiveMQPropertyConversionException {
+ return getFloatProperty(key.toString());
+ }
+
+ @Override
+ public String getStringProperty(SimpleString key) throws ActiveMQPropertyConversionException {
+ return getStringProperty(key.toString());
+ }
+
+ @Override
+ public SimpleString getSimpleStringProperty(SimpleString key) throws ActiveMQPropertyConversionException {
+ return SimpleString.toSimpleString(getStringProperty(key));
+ }
+
+ @Override
+ public byte[] getBytesProperty(SimpleString key) throws ActiveMQPropertyConversionException {
+ return getBytesProperty(key.toString());
+ }
+
+ @Override
+ public Message putStringProperty(SimpleString key, SimpleString value) {
+ return putStringProperty(key.toString(), value.toString());
+ }
+
+ @Override
+ public Message putStringProperty(SimpleString key, String value) {
+ return putStringProperty(key.toString(), value);
+ }
+
+ @Override
+ public int getEncodeSize() {
+ return 0;
+ }
+
+ @Override
+ public Set<SimpleString> getPropertyNames() {
+ try {
+ Set<SimpleString> simpleStrings = new HashSet<>();
+ Enumeration propertyNames = message.getPropertyNames();
+ while (propertyNames.hasMoreElements()) {
+ simpleStrings.add(SimpleString.toSimpleString((String) propertyNames.nextElement()));
+ }
+ return simpleStrings;
+ } catch (JMSException e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public ICoreMessage toCore() {
+ return toCore(null);
+ }
+
+ @Override
+ public ICoreMessage toCore(CoreMessageObjectPools coreMessageObjectPools) {
+ try {
+ return OpenWireMessageConverter.convertAMQToCore(message, coreMessageObjectPools);
+ } catch (Exception e) {
+ throw new ActiveMQPropertyConversionException(e.getMessage());
+ }
+ }
+
+ @Override
+ public int getMemoryEstimate() {
+ return message.getSize();
+ }
+
+ @Override
+ public Object getDuplicateProperty() {
+ return getObjectProperty(Message.HDR_DUPLICATE_DETECTION_ID);
+ }
+
+ @Override
+ public SimpleString getLastValueProperty() {
+ return getSimpleStringProperty(Message.HDR_LAST_VALUE_NAME);
--- End diff --
Done
---
[GitHub] activemq-artemis issue #1793: ARTEMIS-1498: Openwire internal headers should...
Posted by jbertram <gi...@git.apache.org>.
Github user jbertram commented on the issue:
https://github.com/apache/activemq-artemis/pull/1793
@RaiSaurabh, the Travis CI issues should be resolved. Can you please rebase this PR and push -f? Thanks!
---