You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by rg...@apache.org on 2016/07/28 15:20:22 UTC
svn commit: r1754421 - in /qpid/java/trunk:
bdbstore/src/main/java/org/apache/qpid/server/virtualhost/berkeleydb/
broker-core/src/main/java/org/apache/qpid/server/message/internal/
broker-core/src/main/java/org/apache/qpid/server/model/ broker-core/src...
Author: rgodfrey
Date: Thu Jul 28 15:20:22 2016
New Revision: 1754421
URL: http://svn.apache.org/viewvc?rev=1754421&view=rev
Log:
QPID-7366 : Allow publishing of messages using a management operation
Added:
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/model/ManageableMessage.java (with props)
Modified:
qpid/java/trunk/bdbstore/src/main/java/org/apache/qpid/server/virtualhost/berkeleydb/BDBHAReplicaVirtualHostImpl.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/message/internal/InternalMessage.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/message/internal/InternalMessageMetaData.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/model/VirtualHost.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractVirtualHost.java
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/virtualhostnode/RedirectingVirtualHostImpl.java
Modified: qpid/java/trunk/bdbstore/src/main/java/org/apache/qpid/server/virtualhost/berkeleydb/BDBHAReplicaVirtualHostImpl.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/bdbstore/src/main/java/org/apache/qpid/server/virtualhost/berkeleydb/BDBHAReplicaVirtualHostImpl.java?rev=1754421&r1=1754420&r2=1754421&view=diff
==============================================================================
--- qpid/java/trunk/bdbstore/src/main/java/org/apache/qpid/server/virtualhost/berkeleydb/BDBHAReplicaVirtualHostImpl.java (original)
+++ qpid/java/trunk/bdbstore/src/main/java/org/apache/qpid/server/virtualhost/berkeleydb/BDBHAReplicaVirtualHostImpl.java Thu Jul 28 15:20:22 2016
@@ -39,6 +39,7 @@ import org.apache.qpid.server.model.Brok
import org.apache.qpid.server.model.BrokerModel;
import org.apache.qpid.server.model.ConfiguredObject;
import org.apache.qpid.server.model.Connection;
+import org.apache.qpid.server.model.ManageableMessage;
import org.apache.qpid.server.model.ManagedAttributeField;
import org.apache.qpid.server.model.ManagedObject;
import org.apache.qpid.server.model.ManagedObjectFactoryConstructor;
@@ -298,6 +299,13 @@ public class BDBHAReplicaVirtualHostImpl
}
@Override
+ public int publishMessage(final ManageableMessage message)
+ {
+ throwUnsupportedForReplica();
+ return 0;
+ }
+
+ @Override
public MessageSource getAttainedMessageSource(final String name)
{
return null;
Modified: qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/message/internal/InternalMessage.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/message/internal/InternalMessage.java?rev=1754421&r1=1754420&r2=1754421&view=diff
==============================================================================
--- qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/message/internal/InternalMessage.java (original)
+++ qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/message/internal/InternalMessage.java Thu Jul 28 15:20:22 2016
@@ -64,6 +64,7 @@ public class InternalMessage extends Abs
{
super(msg, null);
_contentSize = msg.getMetaData().getContentSize();
+ _header = msg.getMetaData().getHeader();
Collection<QpidByteBuffer> bufs = msg.getContent(0, _contentSize);
try(ObjectInputStream is = new ObjectInputStream(new ByteBufferInputStream(ByteBufferUtils.combine(bufs))))
@@ -117,9 +118,9 @@ public class InternalMessage extends Abs
return _messageBody;
}
- private static InternalMessage createMessage(final MessageStore store,
- final AMQMessageHeader header,
- final Serializable bodyObject, final boolean persistent)
+ public static InternalMessage createMessage(final MessageStore store,
+ final AMQMessageHeader header,
+ final Serializable bodyObject, final boolean persistent)
{
InternalMessageHeader internalHeader;
if(header instanceof InternalMessageHeader)
Modified: qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/message/internal/InternalMessageMetaData.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/message/internal/InternalMessageMetaData.java?rev=1754421&r1=1754420&r2=1754421&view=diff
==============================================================================
--- qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/message/internal/InternalMessageMetaData.java (original)
+++ qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/message/internal/InternalMessageMetaData.java Thu Jul 28 15:20:22 2016
@@ -94,6 +94,11 @@ public class InternalMessageMetaData imp
}
+ InternalMessageHeader getHeader()
+ {
+ return _header;
+ }
+
@Override
public void clearEncodedForm()
{
Added: qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/model/ManageableMessage.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/model/ManageableMessage.java?rev=1754421&view=auto
==============================================================================
--- qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/model/ManageableMessage.java (added)
+++ qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/model/ManageableMessage.java Thu Jul 28 15:20:22 2016
@@ -0,0 +1,54 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.server.model;
+
+import java.util.Date;
+import java.util.Map;
+
+@ManagedAttributeValueType
+public interface ManageableMessage extends ManagedAttributeValue
+{
+ String getAddress();
+
+ boolean isPersistent();
+
+ Date getExpiration();
+
+ String getCorrelationId();
+
+ String getAppId();
+
+ String getMessageId();
+
+ String getMimeType();
+
+ String getEncoding();
+
+ int getPriority();
+
+ Date getNotValidBefore();
+
+ String getReplyTo();
+
+ Map<String,Object> getHeaders();
+
+ Object getContent();
+}
Propchange: qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/model/ManageableMessage.java
------------------------------------------------------------------------------
svn:eol-style = native
Modified: qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/model/VirtualHost.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/model/VirtualHost.java?rev=1754421&r1=1754420&r2=1754421&view=diff
==============================================================================
--- qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/model/VirtualHost.java (original)
+++ qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/model/VirtualHost.java Thu Jul 28 15:20:22 2016
@@ -194,6 +194,9 @@ public interface VirtualHost<X extends V
@ManagedOperation(nonModifying = true)
Connection<?> getConnection(@Param(name="name") String name);
+ @ManagedOperation
+ int publishMessage(@Param(name = "message")ManageableMessage message);
+
void start();
void stop();
Modified: qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractVirtualHost.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractVirtualHost.java?rev=1754421&r1=1754420&r2=1754421&view=diff
==============================================================================
--- qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractVirtualHost.java (original)
+++ qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractVirtualHost.java Thu Jul 28 15:20:22 2016
@@ -23,12 +23,14 @@ package org.apache.qpid.server.virtualho
import static java.util.Collections.newSetFromMap;
import java.io.File;
+import java.io.Serializable;
import java.security.AccessControlContext;
import java.security.Principal;
import java.security.PrivilegedAction;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
+import java.util.Date;
import java.util.EnumSet;
import java.util.HashMap;
import java.util.HashSet;
@@ -45,6 +47,7 @@ import java.util.regex.Pattern;
import java.util.regex.PatternSyntaxException;
import javax.security.auth.Subject;
+import javax.xml.bind.DatatypeConverter;
import com.google.common.base.Function;
import com.google.common.util.concurrent.AsyncFunction;
@@ -67,10 +70,13 @@ import org.apache.qpid.server.logging.Ev
import org.apache.qpid.server.logging.messages.MessageStoreMessages;
import org.apache.qpid.server.logging.messages.VirtualHostMessages;
import org.apache.qpid.server.logging.subjects.MessageStoreLogSubject;
+import org.apache.qpid.server.message.AMQMessageHeader;
+import org.apache.qpid.server.message.InstanceProperties;
import org.apache.qpid.server.message.MessageDestination;
import org.apache.qpid.server.message.MessageNode;
import org.apache.qpid.server.message.MessageSource;
import org.apache.qpid.server.message.ServerMessage;
+import org.apache.qpid.server.message.internal.InternalMessage;
import org.apache.qpid.server.model.*;
import org.apache.qpid.server.model.Connection;
import org.apache.qpid.server.model.port.AmqpPort;
@@ -86,6 +92,7 @@ import org.apache.qpid.server.security.C
import org.apache.qpid.server.security.Result;
import org.apache.qpid.server.security.SecurityToken;
import org.apache.qpid.server.security.access.Operation;
+import org.apache.qpid.server.security.auth.AuthenticatedPrincipal;
import org.apache.qpid.server.stats.StatisticsCounter;
import org.apache.qpid.server.store.ConfiguredObjectRecord;
import org.apache.qpid.server.store.DurableConfigurationStore;
@@ -699,6 +706,81 @@ public abstract class AbstractVirtualHos
}
@Override
+ public int publishMessage(@Param(name = "message") final ManageableMessage message)
+ {
+ MessageDestination destination = getAttainedMessageDestination(message.getAddress());
+ if(destination == null)
+ {
+ destination = getDefaultDestination();
+ }
+
+ final AMQMessageHeader header = new MessageHeaderImpl(message);
+
+ Serializable body = null;
+ Object messageContent = message.getContent();
+ if(messageContent != null)
+ {
+ if(messageContent instanceof Map || messageContent instanceof List)
+ {
+ body = (Serializable)messageContent;
+ }
+ else if(messageContent instanceof String)
+ {
+ if(message.getMimeType() != null || message.getEncoding() != null)
+ {
+ try
+ {
+ body = DatatypeConverter.parseBase64Binary((String)messageContent);
+
+ }
+ catch(IllegalArgumentException e)
+ {
+ body = (String) messageContent;
+ }
+ }
+ else
+ {
+ body = (String) messageContent;
+ }
+ }
+ else
+ {
+ throw new IllegalArgumentException("The message content (if present) can only be a string, map or list");
+ }
+ }
+
+ InternalMessage internalMessage = InternalMessage.createMessage(getMessageStore(), header, body, message.isPersistent());
+ AutoCommitTransaction txn = new AutoCommitTransaction(getMessageStore());
+ final InstanceProperties instanceProperties =
+ new InstanceProperties()
+ {
+ @Override
+ public Object getProperty(final Property prop)
+ {
+ switch (prop)
+ {
+ case EXPIRATION:
+ Date expiration = message.getExpiration();
+ return expiration == null ? 0 : expiration.getTime();
+ case IMMEDIATE:
+ return false;
+ case PERSISTENT:
+ return message.isPersistent();
+ case MANDATORY:
+ return false;
+ case REDELIVERED:
+ return false;
+ default:
+ return null;
+ }
+ }
+ };
+
+ return destination.send(internalMessage, message.getAddress(), instanceProperties, txn, null);
+
+ }
+
+ @Override
protected <C extends ConfiguredObject> ListenableFuture<C> addChildAsync(Class<C> childClass, Map<String, Object> attributes, ConfiguredObject... otherParents)
{
checkVHostStateIsActive();
@@ -1284,6 +1366,118 @@ public abstract class AbstractVirtualHos
return null;
}
+ private static class MessageHeaderImpl implements AMQMessageHeader
+ {
+ private final String _userName;
+ private final long _timestamp;
+ private final ManageableMessage _message;
+
+ public MessageHeaderImpl(final ManageableMessage message)
+ {
+ _message = message;
+ _userName = AuthenticatedPrincipal.getCurrentUser().getName();
+ _timestamp = System.currentTimeMillis();
+ }
+
+ @Override
+ public String getCorrelationId()
+ {
+ return _message.getCorrelationId();
+ }
+
+ @Override
+ public long getExpiration()
+ {
+ Date expiration = _message.getExpiration();
+ return expiration == null ? 0 : expiration.getTime();
+ }
+
+ @Override
+ public String getUserId()
+ {
+ return _userName;
+ }
+
+ @Override
+ public String getAppId()
+ {
+ return null;
+ }
+
+ @Override
+ public String getMessageId()
+ {
+ return _message.getMessageId();
+ }
+
+ @Override
+ public String getMimeType()
+ {
+ return _message.getMimeType();
+ }
+
+ @Override
+ public String getEncoding()
+ {
+ return _message.getEncoding();
+ }
+
+ @Override
+ public byte getPriority()
+ {
+ return (byte) _message.getPriority();
+ }
+
+ @Override
+ public long getTimestamp()
+ {
+ return _timestamp;
+ }
+
+ @Override
+ public long getNotValidBefore()
+ {
+ final Date notValidBefore = _message.getNotValidBefore();
+ return notValidBefore == null ? 0 : notValidBefore.getTime();
+ }
+
+ @Override
+ public String getType()
+ {
+ return null;
+ }
+
+ @Override
+ public String getReplyTo()
+ {
+ return _message.getReplyTo();
+ }
+
+ @Override
+ public Object getHeader(final String name)
+ {
+ return _message.getHeaders().get(name);
+ }
+
+ @Override
+ public boolean containsHeaders(final Set<String> names)
+ {
+ return _message.getHeaders().keySet().containsAll(names);
+ }
+
+ @Override
+ public boolean containsHeader(final String name)
+ {
+ return _message.getHeaders().keySet().contains(name);
+ }
+
+ @Override
+ public Collection<String> getHeaderNames()
+ {
+ return Collections.unmodifiableCollection(_message.getHeaders().keySet());
+ }
+ }
+
private class TargetSizeAssigningListener implements ConfigurationChangeListener
{
@Override
Modified: qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/virtualhostnode/RedirectingVirtualHostImpl.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/virtualhostnode/RedirectingVirtualHostImpl.java?rev=1754421&r1=1754420&r2=1754421&view=diff
==============================================================================
--- qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/virtualhostnode/RedirectingVirtualHostImpl.java (original)
+++ qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/virtualhostnode/RedirectingVirtualHostImpl.java Thu Jul 28 15:20:22 2016
@@ -41,6 +41,7 @@ import org.apache.qpid.server.model.Brok
import org.apache.qpid.server.model.BrokerModel;
import org.apache.qpid.server.model.ConfiguredObject;
import org.apache.qpid.server.model.Connection;
+import org.apache.qpid.server.model.ManageableMessage;
import org.apache.qpid.server.model.ManagedAttributeField;
import org.apache.qpid.server.model.ManagedObject;
import org.apache.qpid.server.model.ManagedObjectFactoryConstructor;
@@ -310,6 +311,13 @@ class RedirectingVirtualHostImpl
}
@Override
+ public int publishMessage(final ManageableMessage message)
+ {
+ throwUnsupportedForRedirector();
+ return 0;
+ }
+
+ @Override
public MessageSource getAttainedMessageSource(final String name)
{
return null;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org