diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/BrokerInterceptor.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/BrokerInterceptor.java index 93007749..5a632342 100644 --- a/broker/src/main/java/org/eclipse/moquette/spi/impl/BrokerInterceptor.java +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/BrokerInterceptor.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2014 The original author or authors + * Copyright (c) 2012-2015 The original author or authors * ------------------------------------------------------ * All rights reserved. This program and the accompanying materials * are made available under the terms of the Eclipse Public License v1.0 @@ -73,11 +73,12 @@ final class BrokerInterceptor implements Interceptor { @Override public void notifyTopicPublished(final PublishMessage msg) { + final PublishMessage copied = new UnmodifiablePublishMessage(msg); for (final InterceptHandler handler : this.handlers) { executor.execute(new Runnable() { @Override public void run() { - handler.onPublish(msg.readOnlyClone()); + handler.onPublish(copied); } }); } diff --git a/broker/src/main/java/org/eclipse/moquette/spi/impl/UnmodifiablePublishMessage.java b/broker/src/main/java/org/eclipse/moquette/spi/impl/UnmodifiablePublishMessage.java new file mode 100644 index 00000000..aaa11d76 --- /dev/null +++ b/broker/src/main/java/org/eclipse/moquette/spi/impl/UnmodifiablePublishMessage.java @@ -0,0 +1,59 @@ +/* + * Copyright (c) 2012-2015 The original author or authors + * ------------------------------------------------------ + * All rights reserved. This program and the accompanying materials + * are made available under the terms of the Eclipse Public License v1.0 + * and Apache License v2.0 which accompanies this distribution. + * + * The Eclipse Public License is available at + * http://www.eclipse.org/legal/epl-v10.html + * + * The Apache License v2.0 is available at + * http://www.opensource.org/licenses/apache2.0.php + * + * You may elect to redistribute this code under either of these licenses. + */ +package org.eclipse.moquette.spi.impl; + +import org.eclipse.moquette.proto.messages.PublishMessage; + +import java.nio.ByteBuffer; + + +/** + * A PublishMessage that is unmodifiable with copy through copy constructor. + * + * @author andrea + */ +final class UnmodifiablePublishMessage extends PublishMessage { + + protected byte m_messageType; + + UnmodifiablePublishMessage(final PublishMessage orig) { + this.m_messageID = orig.getMessageID(); + this.m_topicName = orig.getTopicName(); + if (orig.getPayload() != null) { + this.m_payload = orig.getPayload().asReadOnlyBuffer(); + } + this.m_dupFlag = orig.isDupFlag(); + this.m_qos = orig.getQos(); + this.m_retainFlag = orig.isRetainFlag(); + this.m_remainingLength = orig.getRemainingLength(); + this.m_messageType = orig.getMessageType(); + } + + @Override + public void setTopicName(String topicName) { + throw new UnsupportedOperationException(); + } + + @Override + public void setPayload(ByteBuffer payload) { + throw new UnsupportedOperationException(); + } + + @Override + public void setMessageID(Integer messageID) { + throw new UnsupportedOperationException(); + } +} diff --git a/parser_commons/src/main/java/org/eclipse/moquette/proto/messages/MessageIDMessage.java b/parser_commons/src/main/java/org/eclipse/moquette/proto/messages/MessageIDMessage.java index 150f59d4..940d78d4 100644 --- a/parser_commons/src/main/java/org/eclipse/moquette/proto/messages/MessageIDMessage.java +++ b/parser_commons/src/main/java/org/eclipse/moquette/proto/messages/MessageIDMessage.java @@ -24,7 +24,7 @@ package org.eclipse.moquette.proto.messages; * @author andrea */ public abstract class MessageIDMessage extends AbstractMessage { - private Integer m_messageID; //could be null if Qos is == 0 + protected Integer m_messageID; //could be null if Qos is == 0 public Integer getMessageID() { return m_messageID; diff --git a/parser_commons/src/main/java/org/eclipse/moquette/proto/messages/PublishMessage.java b/parser_commons/src/main/java/org/eclipse/moquette/proto/messages/PublishMessage.java index 42aa23a5..2d1ea237 100644 --- a/parser_commons/src/main/java/org/eclipse/moquette/proto/messages/PublishMessage.java +++ b/parser_commons/src/main/java/org/eclipse/moquette/proto/messages/PublishMessage.java @@ -1,5 +1,5 @@ /* - * Copyright (c) 2012-2014 The original author or authors + * Copyright (c) 2012-2015 The original author or authors * ------------------------------------------------------ * All rights reserved. This program and the accompanying materials * are made available under the terms of the Eclipse Public License v1.0 @@ -23,18 +23,9 @@ import java.nio.ByteBuffer; */ public class PublishMessage extends MessageIDMessage { - private String m_topicName; -// private Integer m_messageID; //could be null if Qos is == 0 - private ByteBuffer m_payload; + protected String m_topicName; + protected ByteBuffer m_payload; - /*public Integer getMessageID() { - return m_messageID; - } - - public void setMessageID(Integer messageID) { - this.m_messageID = messageID; - }*/ - public PublishMessage() { m_messageType = AbstractMessage.PUBLISH; } @@ -55,13 +46,4 @@ public class PublishMessage extends MessageIDMessage { this.m_payload = payload; } - public PublishMessage readOnlyClone() { - try { - final PublishMessage clone = (PublishMessage) super.clone(); - clone.m_payload = this.m_payload.asReadOnlyBuffer(); - return clone; - } catch (CloneNotSupportedException e) { - return null; - } - } }