You are viewing a plain text version of this content. The canonical link for it is here.
Posted to gitbox@activemq.apache.org by "gemmellr (via GitHub)" <gi...@apache.org> on 2023/07/18 09:48:31 UTC
[GitHub] [activemq-artemis] gemmellr commented on a diff in pull request #4555: ARTEMIS-4366 Missing Mirrored ACKs with MULTICAST and subscriptions
gemmellr commented on code in PR #4555:
URL: https://github.com/apache/activemq-artemis/pull/4555#discussion_r1266421706
##########
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/ProtonProtocolManager.java:
##########
@@ -75,7 +75,7 @@ public static String getMirrorAddress(String connectionName) {
// We must use one referenceIDSupplier per server.
// protocol manager is the perfect aggregation for that.
- private ReferenceNodeStore referenceIDSupplier;
+ private ReferenceNodeStoreFactory referenceIDSupplier;
Review Comment:
Change makes the comment above seem inaccurate / contradictory (somewhat similar with the field name). There is now a factory being returned, and so the overall change here seems like it was to make sure there **is not** a single reference ID supplier per server as the comment says, but rather now one per-queue.
##########
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/mirror/ReferenceNodeStoreFactory.java:
##########
@@ -0,0 +1,82 @@
+/*
+ * 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.protocol.amqp.connect.mirror;
+
+import org.apache.activemq.artemis.api.core.Message;
+import org.apache.activemq.artemis.core.server.ActiveMQServer;
+import org.apache.activemq.artemis.core.server.MessageReference;
+import org.apache.activemq.artemis.utils.collections.NodeStore;
+import org.apache.activemq.artemis.utils.collections.NodeStoreFactory;
+
+import static org.apache.activemq.artemis.protocol.amqp.connect.mirror.AMQPMirrorControllerSource.INTERNAL_BROKER_ID_EXTRA_PROPERTY;
+import static org.apache.activemq.artemis.protocol.amqp.connect.mirror.AMQPMirrorControllerSource.INTERNAL_ID_EXTRA_PROPERTY;
+
+public class ReferenceNodeStoreFactory implements NodeStoreFactory<MessageReference> {
+
+ final ActiveMQServer server;
+
+ private final String serverID;
+
+ public ReferenceNodeStoreFactory(ActiveMQServer server) {
+ this.server = server;
+ this.serverID = server.getNodeID().toString();
+
Review Comment:
Superfluous newline
##########
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/ProtonProtocolManager.java:
##########
@@ -125,11 +125,11 @@ public ProtonProtocolManager(ProtonProtocolManagerFactory factory, ActiveMQServe
routingHandler = new AMQPRoutingHandler(server);
}
- public synchronized ReferenceNodeStore getReferenceIDSupplier() {
+ public synchronized ReferenceNodeStoreFactory getReferenceIDSupplier() {
Review Comment:
The method name also seems off now.
##########
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/mirror/ReferenceNodeStoreFactory.java:
##########
@@ -0,0 +1,82 @@
+/*
+ * 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.protocol.amqp.connect.mirror;
+
+import org.apache.activemq.artemis.api.core.Message;
+import org.apache.activemq.artemis.core.server.ActiveMQServer;
+import org.apache.activemq.artemis.core.server.MessageReference;
+import org.apache.activemq.artemis.utils.collections.NodeStore;
+import org.apache.activemq.artemis.utils.collections.NodeStoreFactory;
+
+import static org.apache.activemq.artemis.protocol.amqp.connect.mirror.AMQPMirrorControllerSource.INTERNAL_BROKER_ID_EXTRA_PROPERTY;
+import static org.apache.activemq.artemis.protocol.amqp.connect.mirror.AMQPMirrorControllerSource.INTERNAL_ID_EXTRA_PROPERTY;
+
+public class ReferenceNodeStoreFactory implements NodeStoreFactory<MessageReference> {
+
+ final ActiveMQServer server;
+
+ private final String serverID;
+
+ public ReferenceNodeStoreFactory(ActiveMQServer server) {
+ this.server = server;
+ this.serverID = server.getNodeID().toString();
+
+ }
+
+ @Override
+ public NodeStore<MessageReference> newNodeStore() {
+ return new ReferenceNodeStore(this);
+ }
+
+ public String getDefaultNodeID() {
+ return serverID;
+ }
+
+ public String getServerID(MessageReference element) {
+ return getServerID(element.getMessage());
+ }
+
+
Review Comment:
Superfluous newline
##########
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/mirror/ReferenceNodeStoreFactory.java:
##########
@@ -0,0 +1,82 @@
+/*
+ * 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.protocol.amqp.connect.mirror;
+
+import org.apache.activemq.artemis.api.core.Message;
+import org.apache.activemq.artemis.core.server.ActiveMQServer;
+import org.apache.activemq.artemis.core.server.MessageReference;
+import org.apache.activemq.artemis.utils.collections.NodeStore;
+import org.apache.activemq.artemis.utils.collections.NodeStoreFactory;
+
+import static org.apache.activemq.artemis.protocol.amqp.connect.mirror.AMQPMirrorControllerSource.INTERNAL_BROKER_ID_EXTRA_PROPERTY;
+import static org.apache.activemq.artemis.protocol.amqp.connect.mirror.AMQPMirrorControllerSource.INTERNAL_ID_EXTRA_PROPERTY;
+
+public class ReferenceNodeStoreFactory implements NodeStoreFactory<MessageReference> {
+
+ final ActiveMQServer server;
+
+ private final String serverID;
+
+ public ReferenceNodeStoreFactory(ActiveMQServer server) {
+ this.server = server;
+ this.serverID = server.getNodeID().toString();
+
+ }
+
+ @Override
+ public NodeStore<MessageReference> newNodeStore() {
+ return new ReferenceNodeStore(this);
+ }
+
+ public String getDefaultNodeID() {
+ return serverID;
+ }
+
+ public String getServerID(MessageReference element) {
+ return getServerID(element.getMessage());
+ }
+
+
+ public String getServerID(Message message) {
+ Object nodeID = message.getBrokerProperty(INTERNAL_BROKER_ID_EXTRA_PROPERTY);
+ if (nodeID != null) {
+ return nodeID.toString();
+ } else {
+ // it is important to return null here, as the MirrorSource is expecting it to be null
+ // in the case the nodeID being from the originating server.
+ // don't be tempted to return this.serverID here.
+ return null;
+ }
+ }
+
+ public long getID(MessageReference element) {
+ Message message = element.getMessage();
+ Long id = getID(message);
+ if (id == null) {
+ return element.getMessageID();
+ } else {
+ return id;
+ }
+ }
+
+ private Long getID(Message message) {
+ return (Long)message.getBrokerProperty(INTERNAL_ID_EXTRA_PROPERTY);
+ }
+
+
Review Comment:
Superfluous newline
##########
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/connect/mirror/ReferenceNodeStore.java:
##########
@@ -20,20 +20,16 @@
import io.netty.util.collection.LongObjectHashMap;
import org.apache.activemq.artemis.api.core.Message;
-import org.apache.activemq.artemis.core.server.ActiveMQServer;
import org.apache.activemq.artemis.core.server.MessageReference;
import org.apache.activemq.artemis.utils.collections.NodeStore;
import org.apache.activemq.artemis.utils.collections.LinkedListImpl;
-import static org.apache.activemq.artemis.protocol.amqp.connect.mirror.AMQPMirrorControllerSource.INTERNAL_BROKER_ID_EXTRA_PROPERTY;
-import static org.apache.activemq.artemis.protocol.amqp.connect.mirror.AMQPMirrorControllerSource.INTERNAL_ID_EXTRA_PROPERTY;
-
public class ReferenceNodeStore implements NodeStore<MessageReference> {
- private final String serverID;
+ private final ReferenceNodeStoreFactory factory;
Review Comment:
It seems especially weird for a ReferenceNodeStore to contain the ReferenceNodeStoreFactory that creates ReferenceNodeStore. One which it then only uses to get IDs (which are then really just taken from the message [reference]).
Feels like most or all this usage should be a different interface/ object.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: gitbox-unsubscribe@activemq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org