begin_unit|revision:0.9.5;language:Java;cregit-version:0.0.1
begin_comment
comment|/**  * 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.  */
end_comment

begin_package
package|package
name|org
operator|.
name|apache
operator|.
name|activemq
operator|.
name|usecases
package|;
end_package

begin_import
import|import
name|java
operator|.
name|net
operator|.
name|URI
import|;
end_import

begin_import
import|import
name|java
operator|.
name|util
operator|.
name|Collection
import|;
end_import

begin_import
import|import
name|java
operator|.
name|util
operator|.
name|HashMap
import|;
end_import

begin_import
import|import
name|java
operator|.
name|util
operator|.
name|Iterator
import|;
end_import

begin_import
import|import
name|javax
operator|.
name|jms
operator|.
name|Destination
import|;
end_import

begin_import
import|import
name|javax
operator|.
name|jms
operator|.
name|MessageConsumer
import|;
end_import

begin_import
import|import
name|javax
operator|.
name|jms
operator|.
name|Topic
import|;
end_import

begin_import
import|import
name|junit
operator|.
name|framework
operator|.
name|Test
import|;
end_import

begin_import
import|import
name|org
operator|.
name|apache
operator|.
name|activemq
operator|.
name|JmsMultipleBrokersTestSupport
import|;
end_import

begin_import
import|import
name|org
operator|.
name|apache
operator|.
name|activemq
operator|.
name|broker
operator|.
name|BrokerService
import|;
end_import

begin_import
import|import
name|org
operator|.
name|apache
operator|.
name|activemq
operator|.
name|broker
operator|.
name|region
operator|.
name|policy
operator|.
name|PolicyEntry
import|;
end_import

begin_import
import|import
name|org
operator|.
name|apache
operator|.
name|activemq
operator|.
name|broker
operator|.
name|region
operator|.
name|policy
operator|.
name|PolicyMap
import|;
end_import

begin_import
import|import
name|org
operator|.
name|apache
operator|.
name|activemq
operator|.
name|command
operator|.
name|ActiveMQDestination
import|;
end_import

begin_import
import|import
name|org
operator|.
name|apache
operator|.
name|activemq
operator|.
name|util
operator|.
name|MessageIdList
import|;
end_import

begin_comment
comment|/**  *  */
end_comment

begin_class
specifier|public
class|class
name|ThreeBrokerTopicNetworkTest
extends|extends
name|JmsMultipleBrokersTestSupport
block|{
specifier|protected
specifier|static
specifier|final
name|int
name|MESSAGE_COUNT
init|=
literal|100
decl_stmt|;
specifier|public
name|boolean
name|dynamicOnly
decl_stmt|;
comment|/**      * BrokerA -> BrokerB -> BrokerC      */
specifier|public
name|void
name|testABandBCbrokerNetwork
parameter_list|()
throws|throws
name|Exception
block|{
comment|// Setup broker networks
name|bridgeBrokers
argument_list|(
literal|"BrokerA"
argument_list|,
literal|"BrokerB"
argument_list|)
expr_stmt|;
name|bridgeBrokers
argument_list|(
literal|"BrokerB"
argument_list|,
literal|"BrokerC"
argument_list|)
expr_stmt|;
name|startAllBrokers
argument_list|()
expr_stmt|;
comment|// Setup destination
name|Destination
name|dest
init|=
name|createDestination
argument_list|(
literal|"TEST.FOO"
argument_list|,
literal|true
argument_list|)
decl_stmt|;
comment|// Setup consumers
name|MessageConsumer
name|clientA
init|=
name|createConsumer
argument_list|(
literal|"BrokerA"
argument_list|,
name|dest
argument_list|)
decl_stmt|;
name|MessageConsumer
name|clientB
init|=
name|createConsumer
argument_list|(
literal|"BrokerB"
argument_list|,
name|dest
argument_list|)
decl_stmt|;
name|MessageConsumer
name|clientC
init|=
name|createConsumer
argument_list|(
literal|"BrokerC"
argument_list|,
name|dest
argument_list|)
decl_stmt|;
comment|//let consumers propagate around the network
name|Thread
operator|.
name|sleep
argument_list|(
literal|2000
argument_list|)
expr_stmt|;
comment|// Send messages
name|sendMessages
argument_list|(
literal|"BrokerA"
argument_list|,
name|dest
argument_list|,
name|MESSAGE_COUNT
argument_list|)
expr_stmt|;
name|sendMessages
argument_list|(
literal|"BrokerB"
argument_list|,
name|dest
argument_list|,
name|MESSAGE_COUNT
argument_list|)
expr_stmt|;
name|sendMessages
argument_list|(
literal|"BrokerC"
argument_list|,
name|dest
argument_list|,
name|MESSAGE_COUNT
argument_list|)
expr_stmt|;
comment|// Get message count
name|MessageIdList
name|msgsA
init|=
name|getConsumerMessages
argument_list|(
literal|"BrokerA"
argument_list|,
name|clientA
argument_list|)
decl_stmt|;
name|MessageIdList
name|msgsB
init|=
name|getConsumerMessages
argument_list|(
literal|"BrokerB"
argument_list|,
name|clientB
argument_list|)
decl_stmt|;
name|MessageIdList
name|msgsC
init|=
name|getConsumerMessages
argument_list|(
literal|"BrokerC"
argument_list|,
name|clientC
argument_list|)
decl_stmt|;
name|msgsA
operator|.
name|waitForMessagesToArrive
argument_list|(
name|MESSAGE_COUNT
argument_list|)
expr_stmt|;
name|msgsB
operator|.
name|waitForMessagesToArrive
argument_list|(
name|MESSAGE_COUNT
operator|*
literal|2
argument_list|)
expr_stmt|;
name|msgsC
operator|.
name|waitForMessagesToArrive
argument_list|(
name|MESSAGE_COUNT
operator|*
literal|2
argument_list|)
expr_stmt|;
name|assertEquals
argument_list|(
name|MESSAGE_COUNT
argument_list|,
name|msgsA
operator|.
name|getMessageCount
argument_list|()
argument_list|)
expr_stmt|;
name|assertEquals
argument_list|(
name|MESSAGE_COUNT
operator|*
literal|2
argument_list|,
name|msgsB
operator|.
name|getMessageCount
argument_list|()
argument_list|)
expr_stmt|;
name|assertEquals
argument_list|(
name|MESSAGE_COUNT
operator|*
literal|2
argument_list|,
name|msgsC
operator|.
name|getMessageCount
argument_list|()
argument_list|)
expr_stmt|;
name|assertEquals
argument_list|(
literal|"Correct forwards from A"
argument_list|,
name|MESSAGE_COUNT
argument_list|,
name|brokers
operator|.
name|get
argument_list|(
literal|"BrokerA"
argument_list|)
operator|.
name|broker
operator|.
name|getDestination
argument_list|(
name|ActiveMQDestination
operator|.
name|transform
argument_list|(
name|dest
argument_list|)
argument_list|)
operator|.
name|getDestinationStatistics
argument_list|()
operator|.
name|getForwards
argument_list|()
operator|.
name|getCount
argument_list|()
argument_list|)
expr_stmt|;
block|}
specifier|public
name|void
name|initCombosForTestABandBCbrokerNetworkWithSelectors
parameter_list|()
block|{
name|addCombinationValues
argument_list|(
literal|"dynamicOnly"
argument_list|,
operator|new
name|Object
index|[]
block|{
literal|true
block|,
literal|false
block|}
argument_list|)
expr_stmt|;
block|}
comment|/**      * BrokerA -> BrokerB -> BrokerC      */
specifier|public
name|void
name|testABandBCbrokerNetworkWithSelectors
parameter_list|()
throws|throws
name|Exception
block|{
comment|// Setup broker networks
name|bridgeBrokers
argument_list|(
literal|"BrokerA"
argument_list|,
literal|"BrokerB"
argument_list|,
name|dynamicOnly
argument_list|,
literal|2
argument_list|,
literal|true
argument_list|)
expr_stmt|;
name|bridgeBrokers
argument_list|(
literal|"BrokerB"
argument_list|,
literal|"BrokerC"
argument_list|,
name|dynamicOnly
argument_list|,
literal|2
argument_list|,
literal|true
argument_list|)
expr_stmt|;
name|startAllBrokers
argument_list|()
expr_stmt|;
comment|// Setup destination
name|Destination
name|dest
init|=
name|createDestination
argument_list|(
literal|"TEST.FOO"
argument_list|,
literal|true
argument_list|)
decl_stmt|;
comment|// Setup consumers
name|MessageConsumer
name|clientA
init|=
name|createConsumer
argument_list|(
literal|"BrokerC"
argument_list|,
name|dest
argument_list|,
literal|"dummy = 33"
argument_list|)
decl_stmt|;
name|MessageConsumer
name|clientB
init|=
name|createConsumer
argument_list|(
literal|"BrokerC"
argument_list|,
name|dest
argument_list|,
literal|"dummy> 30"
argument_list|)
decl_stmt|;
name|MessageConsumer
name|clientC
init|=
name|createConsumer
argument_list|(
literal|"BrokerC"
argument_list|,
name|dest
argument_list|,
literal|"dummy = 34"
argument_list|)
decl_stmt|;
comment|// let consumers propogate around the network
name|Thread
operator|.
name|sleep
argument_list|(
literal|2000
argument_list|)
expr_stmt|;
comment|// Send messages
comment|// Send messages for broker A
name|HashMap
argument_list|<
name|String
argument_list|,
name|Object
argument_list|>
name|props
init|=
operator|new
name|HashMap
argument_list|<
name|String
argument_list|,
name|Object
argument_list|>
argument_list|()
decl_stmt|;
name|props
operator|.
name|put
argument_list|(
literal|"dummy"
argument_list|,
literal|33
argument_list|)
expr_stmt|;
name|sendMessages
argument_list|(
literal|"BrokerA"
argument_list|,
name|dest
argument_list|,
name|MESSAGE_COUNT
argument_list|,
name|props
argument_list|)
expr_stmt|;
name|props
operator|.
name|put
argument_list|(
literal|"dummy"
argument_list|,
literal|34
argument_list|)
expr_stmt|;
name|sendMessages
argument_list|(
literal|"BrokerA"
argument_list|,
name|dest
argument_list|,
name|MESSAGE_COUNT
operator|*
literal|2
argument_list|,
name|props
argument_list|)
expr_stmt|;
comment|// Get message count
name|MessageIdList
name|msgsA
init|=
name|getConsumerMessages
argument_list|(
literal|"BrokerC"
argument_list|,
name|clientA
argument_list|)
decl_stmt|;
name|MessageIdList
name|msgsB
init|=
name|getConsumerMessages
argument_list|(
literal|"BrokerC"
argument_list|,
name|clientB
argument_list|)
decl_stmt|;
name|MessageIdList
name|msgsC
init|=
name|getConsumerMessages
argument_list|(
literal|"BrokerC"
argument_list|,
name|clientC
argument_list|)
decl_stmt|;
name|msgsA
operator|.
name|waitForMessagesToArrive
argument_list|(
name|MESSAGE_COUNT
argument_list|)
expr_stmt|;
name|msgsB
operator|.
name|waitForMessagesToArrive
argument_list|(
name|MESSAGE_COUNT
operator|*
literal|3
argument_list|)
expr_stmt|;
name|msgsC
operator|.
name|waitForMessagesToArrive
argument_list|(
name|MESSAGE_COUNT
operator|*
literal|2
argument_list|)
expr_stmt|;
name|assertEquals
argument_list|(
name|MESSAGE_COUNT
argument_list|,
name|msgsA
operator|.
name|getMessageCount
argument_list|()
argument_list|)
expr_stmt|;
name|assertEquals
argument_list|(
name|MESSAGE_COUNT
operator|*
literal|3
argument_list|,
name|msgsB
operator|.
name|getMessageCount
argument_list|()
argument_list|)
expr_stmt|;
name|assertEquals
argument_list|(
name|MESSAGE_COUNT
operator|*
literal|2
argument_list|,
name|msgsC
operator|.
name|getMessageCount
argument_list|()
argument_list|)
expr_stmt|;
block|}
comment|/**      * BrokerA<- BrokerB -> BrokerC      */
specifier|public
name|void
name|testBAandBCbrokerNetwork
parameter_list|()
throws|throws
name|Exception
block|{
comment|// Setup broker networks
name|bridgeBrokers
argument_list|(
literal|"BrokerB"
argument_list|,
literal|"BrokerA"
argument_list|)
expr_stmt|;
name|bridgeBrokers
argument_list|(
literal|"BrokerB"
argument_list|,
literal|"BrokerC"
argument_list|)
expr_stmt|;
name|startAllBrokers
argument_list|()
expr_stmt|;
comment|// Setup destination
name|Destination
name|dest
init|=
name|createDestination
argument_list|(
literal|"TEST.FOO"
argument_list|,
literal|true
argument_list|)
decl_stmt|;
comment|// Setup consumers
name|MessageConsumer
name|clientA
init|=
name|createConsumer
argument_list|(
literal|"BrokerA"
argument_list|,
name|dest
argument_list|)
decl_stmt|;
name|MessageConsumer
name|clientB
init|=
name|createConsumer
argument_list|(
literal|"BrokerB"
argument_list|,
name|dest
argument_list|)
decl_stmt|;
name|MessageConsumer
name|clientC
init|=
name|createConsumer
argument_list|(
literal|"BrokerC"
argument_list|,
name|dest
argument_list|)
decl_stmt|;
comment|//let consumers propagate around the network
name|Thread
operator|.
name|sleep
argument_list|(
literal|2000
argument_list|)
expr_stmt|;
comment|// Send messages
name|sendMessages
argument_list|(
literal|"BrokerA"
argument_list|,
name|dest
argument_list|,
name|MESSAGE_COUNT
argument_list|)
expr_stmt|;
name|sendMessages
argument_list|(
literal|"BrokerB"
argument_list|,
name|dest
argument_list|,
name|MESSAGE_COUNT
argument_list|)
expr_stmt|;
name|sendMessages
argument_list|(
literal|"BrokerC"
argument_list|,
name|dest
argument_list|,
name|MESSAGE_COUNT
argument_list|)
expr_stmt|;
comment|// Get message count
name|MessageIdList
name|msgsA
init|=
name|getConsumerMessages
argument_list|(
literal|"BrokerA"
argument_list|,
name|clientA
argument_list|)
decl_stmt|;
name|MessageIdList
name|msgsB
init|=
name|getConsumerMessages
argument_list|(
literal|"BrokerB"
argument_list|,
name|clientB
argument_list|)
decl_stmt|;
name|MessageIdList
name|msgsC
init|=
name|getConsumerMessages
argument_list|(
literal|"BrokerC"
argument_list|,
name|clientC
argument_list|)
decl_stmt|;
name|msgsA
operator|.
name|waitForMessagesToArrive
argument_list|(
name|MESSAGE_COUNT
operator|*
literal|2
argument_list|)
expr_stmt|;
name|msgsB
operator|.
name|waitForMessagesToArrive
argument_list|(
name|MESSAGE_COUNT
argument_list|)
expr_stmt|;
name|msgsC
operator|.
name|waitForMessagesToArrive
argument_list|(
name|MESSAGE_COUNT
operator|*
literal|2
argument_list|)
expr_stmt|;
name|assertEquals
argument_list|(
name|MESSAGE_COUNT
operator|*
literal|2
argument_list|,
name|msgsA
operator|.
name|getMessageCount
argument_list|()
argument_list|)
expr_stmt|;
name|assertEquals
argument_list|(
name|MESSAGE_COUNT
argument_list|,
name|msgsB
operator|.
name|getMessageCount
argument_list|()
argument_list|)
expr_stmt|;
name|assertEquals
argument_list|(
name|MESSAGE_COUNT
operator|*
literal|2
argument_list|,
name|msgsC
operator|.
name|getMessageCount
argument_list|()
argument_list|)
expr_stmt|;
block|}
comment|/**      * BrokerA -> BrokerB<- BrokerC      */
specifier|public
name|void
name|testABandCBbrokerNetwork
parameter_list|()
throws|throws
name|Exception
block|{
comment|// Setup broker networks
name|bridgeBrokers
argument_list|(
literal|"BrokerA"
argument_list|,
literal|"BrokerB"
argument_list|)
expr_stmt|;
name|bridgeBrokers
argument_list|(
literal|"BrokerC"
argument_list|,
literal|"BrokerB"
argument_list|)
expr_stmt|;
name|startAllBrokers
argument_list|()
expr_stmt|;
comment|// Setup destination
name|Destination
name|dest
init|=
name|createDestination
argument_list|(
literal|"TEST.FOO"
argument_list|,
literal|true
argument_list|)
decl_stmt|;
comment|// Setup consumers
name|MessageConsumer
name|clientA
init|=
name|createConsumer
argument_list|(
literal|"BrokerA"
argument_list|,
name|dest
argument_list|)
decl_stmt|;
name|MessageConsumer
name|clientB
init|=
name|createConsumer
argument_list|(
literal|"BrokerB"
argument_list|,
name|dest
argument_list|)
decl_stmt|;
name|MessageConsumer
name|clientC
init|=
name|createConsumer
argument_list|(
literal|"BrokerC"
argument_list|,
name|dest
argument_list|)
decl_stmt|;
comment|//let consumers propagate around the network
name|Thread
operator|.
name|sleep
argument_list|(
literal|2000
argument_list|)
expr_stmt|;
comment|// Send messages
name|sendMessages
argument_list|(
literal|"BrokerA"
argument_list|,
name|dest
argument_list|,
name|MESSAGE_COUNT
argument_list|)
expr_stmt|;
name|sendMessages
argument_list|(
literal|"BrokerB"
argument_list|,
name|dest
argument_list|,
name|MESSAGE_COUNT
argument_list|)
expr_stmt|;
name|sendMessages
argument_list|(
literal|"BrokerC"
argument_list|,
name|dest
argument_list|,
name|MESSAGE_COUNT
argument_list|)
expr_stmt|;
comment|// Get message count
name|MessageIdList
name|msgsA
init|=
name|getConsumerMessages
argument_list|(
literal|"BrokerA"
argument_list|,
name|clientA
argument_list|)
decl_stmt|;
name|MessageIdList
name|msgsB
init|=
name|getConsumerMessages
argument_list|(
literal|"BrokerB"
argument_list|,
name|clientB
argument_list|)
decl_stmt|;
name|MessageIdList
name|msgsC
init|=
name|getConsumerMessages
argument_list|(
literal|"BrokerC"
argument_list|,
name|clientC
argument_list|)
decl_stmt|;
name|msgsA
operator|.
name|waitForMessagesToArrive
argument_list|(
name|MESSAGE_COUNT
argument_list|)
expr_stmt|;
name|msgsB
operator|.
name|waitForMessagesToArrive
argument_list|(
name|MESSAGE_COUNT
operator|*
literal|3
argument_list|)
expr_stmt|;
name|msgsC
operator|.
name|waitForMessagesToArrive
argument_list|(
name|MESSAGE_COUNT
argument_list|)
expr_stmt|;
name|assertEquals
argument_list|(
name|MESSAGE_COUNT
argument_list|,
name|msgsA
operator|.
name|getMessageCount
argument_list|()
argument_list|)
expr_stmt|;
name|assertEquals
argument_list|(
name|MESSAGE_COUNT
operator|*
literal|3
argument_list|,
name|msgsB
operator|.
name|getMessageCount
argument_list|()
argument_list|)
expr_stmt|;
name|assertEquals
argument_list|(
name|MESSAGE_COUNT
argument_list|,
name|msgsC
operator|.
name|getMessageCount
argument_list|()
argument_list|)
expr_stmt|;
block|}
comment|/**      * BrokerA<-> BrokerB<-> BrokerC      */
specifier|public
name|void
name|testAllConnectedBrokerNetwork
parameter_list|()
throws|throws
name|Exception
block|{
comment|// Setup broker networks
name|bridgeBrokers
argument_list|(
literal|"BrokerA"
argument_list|,
literal|"BrokerB"
argument_list|)
expr_stmt|;
name|bridgeBrokers
argument_list|(
literal|"BrokerB"
argument_list|,
literal|"BrokerA"
argument_list|)
expr_stmt|;
name|bridgeBrokers
argument_list|(
literal|"BrokerB"
argument_list|,
literal|"BrokerC"
argument_list|)
expr_stmt|;
name|bridgeBrokers
argument_list|(
literal|"BrokerC"
argument_list|,
literal|"BrokerB"
argument_list|)
expr_stmt|;
name|bridgeBrokers
argument_list|(
literal|"BrokerA"
argument_list|,
literal|"BrokerC"
argument_list|)
expr_stmt|;
name|bridgeBrokers
argument_list|(
literal|"BrokerC"
argument_list|,
literal|"BrokerA"
argument_list|)
expr_stmt|;
name|startAllBrokers
argument_list|()
expr_stmt|;
comment|// Setup destination
name|Destination
name|dest
init|=
name|createDestination
argument_list|(
literal|"TEST.FOO"
argument_list|,
literal|true
argument_list|)
decl_stmt|;
comment|// Setup consumers
name|MessageConsumer
name|clientA
init|=
name|createConsumer
argument_list|(
literal|"BrokerA"
argument_list|,
name|dest
argument_list|)
decl_stmt|;
name|MessageConsumer
name|clientB
init|=
name|createConsumer
argument_list|(
literal|"BrokerB"
argument_list|,
name|dest
argument_list|)
decl_stmt|;
name|MessageConsumer
name|clientC
init|=
name|createConsumer
argument_list|(
literal|"BrokerC"
argument_list|,
name|dest
argument_list|)
decl_stmt|;
comment|//let consumers propogate around the network
name|Thread
operator|.
name|sleep
argument_list|(
literal|2000
argument_list|)
expr_stmt|;
comment|// Send messages
name|sendMessages
argument_list|(
literal|"BrokerA"
argument_list|,
name|dest
argument_list|,
name|MESSAGE_COUNT
argument_list|)
expr_stmt|;
name|sendMessages
argument_list|(
literal|"BrokerB"
argument_list|,
name|dest
argument_list|,
name|MESSAGE_COUNT
argument_list|)
expr_stmt|;
name|sendMessages
argument_list|(
literal|"BrokerC"
argument_list|,
name|dest
argument_list|,
name|MESSAGE_COUNT
argument_list|)
expr_stmt|;
comment|// Get message count
name|MessageIdList
name|msgsA
init|=
name|getConsumerMessages
argument_list|(
literal|"BrokerA"
argument_list|,
name|clientA
argument_list|)
decl_stmt|;
name|MessageIdList
name|msgsB
init|=
name|getConsumerMessages
argument_list|(
literal|"BrokerB"
argument_list|,
name|clientB
argument_list|)
decl_stmt|;
name|MessageIdList
name|msgsC
init|=
name|getConsumerMessages
argument_list|(
literal|"BrokerC"
argument_list|,
name|clientC
argument_list|)
decl_stmt|;
name|msgsA
operator|.
name|waitForMessagesToArrive
argument_list|(
name|MESSAGE_COUNT
operator|*
literal|3
argument_list|)
expr_stmt|;
name|msgsB
operator|.
name|waitForMessagesToArrive
argument_list|(
name|MESSAGE_COUNT
operator|*
literal|3
argument_list|)
expr_stmt|;
name|msgsC
operator|.
name|waitForMessagesToArrive
argument_list|(
name|MESSAGE_COUNT
operator|*
literal|3
argument_list|)
expr_stmt|;
name|assertEquals
argument_list|(
name|MESSAGE_COUNT
operator|*
literal|3
argument_list|,
name|msgsA
operator|.
name|getMessageCount
argument_list|()
argument_list|)
expr_stmt|;
name|assertEquals
argument_list|(
name|MESSAGE_COUNT
operator|*
literal|3
argument_list|,
name|msgsB
operator|.
name|getMessageCount
argument_list|()
argument_list|)
expr_stmt|;
name|assertEquals
argument_list|(
name|MESSAGE_COUNT
operator|*
literal|3
argument_list|,
name|msgsC
operator|.
name|getMessageCount
argument_list|()
argument_list|)
expr_stmt|;
block|}
specifier|public
name|void
name|testAllConnectedBrokerNetworkSingleProducerTTL
parameter_list|()
throws|throws
name|Exception
block|{
comment|// duplicates are expected with ttl of 2 as each broker is connected to the next
comment|// but the dups are suppressed by the store and now also by the topic sub when enableAudit
comment|// default (true) is present in a matching destination policy entry
name|int
name|networkTTL
init|=
literal|2
decl_stmt|;
name|boolean
name|conduitSubs
init|=
literal|true
decl_stmt|;
comment|// Setup ring broker networks
name|bridgeBrokers
argument_list|(
literal|"BrokerA"
argument_list|,
literal|"BrokerB"
argument_list|,
name|dynamicOnly
argument_list|,
name|networkTTL
argument_list|,
name|conduitSubs
argument_list|)
expr_stmt|;
name|bridgeBrokers
argument_list|(
literal|"BrokerB"
argument_list|,
literal|"BrokerA"
argument_list|,
name|dynamicOnly
argument_list|,
name|networkTTL
argument_list|,
name|conduitSubs
argument_list|)
expr_stmt|;
name|bridgeBrokers
argument_list|(
literal|"BrokerB"
argument_list|,
literal|"BrokerC"
argument_list|,
name|dynamicOnly
argument_list|,
name|networkTTL
argument_list|,
name|conduitSubs
argument_list|)
expr_stmt|;
name|bridgeBrokers
argument_list|(
literal|"BrokerC"
argument_list|,
literal|"BrokerB"
argument_list|,
name|dynamicOnly
argument_list|,
name|networkTTL
argument_list|,
name|conduitSubs
argument_list|)
expr_stmt|;
name|bridgeBrokers
argument_list|(
literal|"BrokerA"
argument_list|,
literal|"BrokerC"
argument_list|,
name|dynamicOnly
argument_list|,
name|networkTTL
argument_list|,
name|conduitSubs
argument_list|)
expr_stmt|;
name|bridgeBrokers
argument_list|(
literal|"BrokerC"
argument_list|,
literal|"BrokerA"
argument_list|,
name|dynamicOnly
argument_list|,
name|networkTTL
argument_list|,
name|conduitSubs
argument_list|)
expr_stmt|;
name|PolicyMap
name|policyMap
init|=
operator|new
name|PolicyMap
argument_list|()
decl_stmt|;
comment|// enable audit is on by default just need to give it matching policy entry
comment|// so it will be applied to the topic subscription
name|policyMap
operator|.
name|setDefaultEntry
argument_list|(
operator|new
name|PolicyEntry
argument_list|()
argument_list|)
expr_stmt|;
name|Collection
argument_list|<
name|BrokerItem
argument_list|>
name|brokerList
init|=
name|brokers
operator|.
name|values
argument_list|()
decl_stmt|;
for|for
control|(
name|Iterator
argument_list|<
name|BrokerItem
argument_list|>
name|i
init|=
name|brokerList
operator|.
name|iterator
argument_list|()
init|;
name|i
operator|.
name|hasNext
argument_list|()
condition|;
control|)
block|{
name|BrokerService
name|broker
init|=
name|i
operator|.
name|next
argument_list|()
operator|.
name|broker
decl_stmt|;
name|broker
operator|.
name|setDestinationPolicy
argument_list|(
name|policyMap
argument_list|)
expr_stmt|;
name|broker
operator|.
name|setDeleteAllMessagesOnStartup
argument_list|(
literal|true
argument_list|)
expr_stmt|;
block|}
name|startAllBrokers
argument_list|()
expr_stmt|;
comment|// Setup destination
name|Destination
name|dest
init|=
name|createDestination
argument_list|(
literal|"TEST.FOO"
argument_list|,
literal|true
argument_list|)
decl_stmt|;
comment|// Setup consumers
name|MessageConsumer
name|clientA
init|=
name|createConsumer
argument_list|(
literal|"BrokerA"
argument_list|,
name|dest
argument_list|)
decl_stmt|;
name|MessageConsumer
name|clientB
init|=
name|createConsumer
argument_list|(
literal|"BrokerB"
argument_list|,
name|dest
argument_list|)
decl_stmt|;
name|MessageConsumer
name|clientC
init|=
name|createConsumer
argument_list|(
literal|"BrokerC"
argument_list|,
name|dest
argument_list|)
decl_stmt|;
comment|//let consumers propogate around the network
name|Thread
operator|.
name|sleep
argument_list|(
literal|2000
argument_list|)
expr_stmt|;
comment|// Send messages
name|sendMessages
argument_list|(
literal|"BrokerA"
argument_list|,
name|dest
argument_list|,
literal|1
argument_list|)
expr_stmt|;
comment|// Get message count
name|MessageIdList
name|msgsA
init|=
name|getConsumerMessages
argument_list|(
literal|"BrokerA"
argument_list|,
name|clientA
argument_list|)
decl_stmt|;
name|MessageIdList
name|msgsB
init|=
name|getConsumerMessages
argument_list|(
literal|"BrokerB"
argument_list|,
name|clientB
argument_list|)
decl_stmt|;
name|MessageIdList
name|msgsC
init|=
name|getConsumerMessages
argument_list|(
literal|"BrokerC"
argument_list|,
name|clientC
argument_list|)
decl_stmt|;
name|msgsA
operator|.
name|waitForMessagesToArrive
argument_list|(
literal|1
argument_list|)
expr_stmt|;
name|msgsB
operator|.
name|waitForMessagesToArrive
argument_list|(
literal|1
argument_list|)
expr_stmt|;
name|msgsC
operator|.
name|waitForMessagesToArrive
argument_list|(
literal|1
argument_list|)
expr_stmt|;
comment|// ensure we don't get any more messages
name|Thread
operator|.
name|sleep
argument_list|(
literal|2000
argument_list|)
expr_stmt|;
name|assertEquals
argument_list|(
literal|1
argument_list|,
name|msgsA
operator|.
name|getMessageCount
argument_list|()
argument_list|)
expr_stmt|;
name|assertEquals
argument_list|(
literal|1
argument_list|,
name|msgsB
operator|.
name|getMessageCount
argument_list|()
argument_list|)
expr_stmt|;
name|assertEquals
argument_list|(
literal|1
argument_list|,
name|msgsC
operator|.
name|getMessageCount
argument_list|()
argument_list|)
expr_stmt|;
block|}
specifier|public
name|void
name|testAllConnectedBrokerNetworkDurableSubTTL
parameter_list|()
throws|throws
name|Exception
block|{
name|int
name|networkTTL
init|=
literal|2
decl_stmt|;
name|boolean
name|conduitSubs
init|=
literal|true
decl_stmt|;
comment|// Setup ring broker network
name|bridgeBrokers
argument_list|(
literal|"BrokerA"
argument_list|,
literal|"BrokerB"
argument_list|,
name|dynamicOnly
argument_list|,
name|networkTTL
argument_list|,
name|conduitSubs
argument_list|)
expr_stmt|;
name|bridgeBrokers
argument_list|(
literal|"BrokerB"
argument_list|,
literal|"BrokerA"
argument_list|,
name|dynamicOnly
argument_list|,
name|networkTTL
argument_list|,
name|conduitSubs
argument_list|)
expr_stmt|;
name|bridgeBrokers
argument_list|(
literal|"BrokerB"
argument_list|,
literal|"BrokerC"
argument_list|,
name|dynamicOnly
argument_list|,
name|networkTTL
argument_list|,
name|conduitSubs
argument_list|)
expr_stmt|;
name|bridgeBrokers
argument_list|(
literal|"BrokerC"
argument_list|,
literal|"BrokerB"
argument_list|,
name|dynamicOnly
argument_list|,
name|networkTTL
argument_list|,
name|conduitSubs
argument_list|)
expr_stmt|;
name|bridgeBrokers
argument_list|(
literal|"BrokerA"
argument_list|,
literal|"BrokerC"
argument_list|,
name|dynamicOnly
argument_list|,
name|networkTTL
argument_list|,
name|conduitSubs
argument_list|)
expr_stmt|;
name|bridgeBrokers
argument_list|(
literal|"BrokerC"
argument_list|,
literal|"BrokerA"
argument_list|,
name|dynamicOnly
argument_list|,
name|networkTTL
argument_list|,
name|conduitSubs
argument_list|)
expr_stmt|;
name|startAllBrokers
argument_list|()
expr_stmt|;
comment|// Setup destination
name|Destination
name|dest
init|=
name|createDestination
argument_list|(
literal|"TEST.FOO"
argument_list|,
literal|true
argument_list|)
decl_stmt|;
comment|// Setup consumers
name|MessageConsumer
name|clientA
init|=
name|createDurableSubscriber
argument_list|(
literal|"BrokerA"
argument_list|,
operator|(
name|Topic
operator|)
name|dest
argument_list|,
literal|"clientA"
argument_list|)
decl_stmt|;
name|MessageConsumer
name|clientB
init|=
name|createDurableSubscriber
argument_list|(
literal|"BrokerB"
argument_list|,
operator|(
name|Topic
operator|)
name|dest
argument_list|,
literal|"clientB"
argument_list|)
decl_stmt|;
name|MessageConsumer
name|clientC
init|=
name|createDurableSubscriber
argument_list|(
literal|"BrokerC"
argument_list|,
operator|(
name|Topic
operator|)
name|dest
argument_list|,
literal|"clientC"
argument_list|)
decl_stmt|;
comment|//let consumers propogate around the network
name|Thread
operator|.
name|sleep
argument_list|(
literal|2000
argument_list|)
expr_stmt|;
comment|// Send messages
name|sendMessages
argument_list|(
literal|"BrokerA"
argument_list|,
name|dest
argument_list|,
literal|1
argument_list|)
expr_stmt|;
comment|// Get message count
name|MessageIdList
name|msgsA
init|=
name|getConsumerMessages
argument_list|(
literal|"BrokerA"
argument_list|,
name|clientA
argument_list|)
decl_stmt|;
name|MessageIdList
name|msgsB
init|=
name|getConsumerMessages
argument_list|(
literal|"BrokerB"
argument_list|,
name|clientB
argument_list|)
decl_stmt|;
name|MessageIdList
name|msgsC
init|=
name|getConsumerMessages
argument_list|(
literal|"BrokerC"
argument_list|,
name|clientC
argument_list|)
decl_stmt|;
name|msgsA
operator|.
name|waitForMessagesToArrive
argument_list|(
literal|1
argument_list|)
expr_stmt|;
name|msgsB
operator|.
name|waitForMessagesToArrive
argument_list|(
literal|1
argument_list|)
expr_stmt|;
name|msgsC
operator|.
name|waitForMessagesToArrive
argument_list|(
literal|1
argument_list|)
expr_stmt|;
comment|// ensure we don't get any more messages
name|Thread
operator|.
name|sleep
argument_list|(
literal|2000
argument_list|)
expr_stmt|;
name|assertEquals
argument_list|(
literal|1
argument_list|,
name|msgsA
operator|.
name|getMessageCount
argument_list|()
argument_list|)
expr_stmt|;
name|assertEquals
argument_list|(
literal|1
argument_list|,
name|msgsB
operator|.
name|getMessageCount
argument_list|()
argument_list|)
expr_stmt|;
name|assertEquals
argument_list|(
literal|1
argument_list|,
name|msgsC
operator|.
name|getMessageCount
argument_list|()
argument_list|)
expr_stmt|;
block|}
comment|/**      * BrokerA<-> BrokerB<-> BrokerC      */
specifier|public
name|void
name|testAllConnectedUsingMulticast
parameter_list|()
throws|throws
name|Exception
block|{
comment|// Setup broker networks
name|bridgeAllBrokers
argument_list|()
expr_stmt|;
name|startAllBrokers
argument_list|()
expr_stmt|;
comment|// Setup destination
name|Destination
name|dest
init|=
name|createDestination
argument_list|(
literal|"TEST.FOO"
argument_list|,
literal|true
argument_list|)
decl_stmt|;
comment|// Setup consumers
name|MessageConsumer
name|clientA
init|=
name|createConsumer
argument_list|(
literal|"BrokerA"
argument_list|,
name|dest
argument_list|)
decl_stmt|;
name|MessageConsumer
name|clientB
init|=
name|createConsumer
argument_list|(
literal|"BrokerB"
argument_list|,
name|dest
argument_list|)
decl_stmt|;
name|MessageConsumer
name|clientC
init|=
name|createConsumer
argument_list|(
literal|"BrokerC"
argument_list|,
name|dest
argument_list|)
decl_stmt|;
comment|//let consumers propogate around the network
name|Thread
operator|.
name|sleep
argument_list|(
literal|2000
argument_list|)
expr_stmt|;
comment|// Send messages
name|sendMessages
argument_list|(
literal|"BrokerA"
argument_list|,
name|dest
argument_list|,
name|MESSAGE_COUNT
argument_list|)
expr_stmt|;
name|sendMessages
argument_list|(
literal|"BrokerB"
argument_list|,
name|dest
argument_list|,
name|MESSAGE_COUNT
argument_list|)
expr_stmt|;
name|sendMessages
argument_list|(
literal|"BrokerC"
argument_list|,
name|dest
argument_list|,
name|MESSAGE_COUNT
argument_list|)
expr_stmt|;
comment|// Get message count
name|MessageIdList
name|msgsA
init|=
name|getConsumerMessages
argument_list|(
literal|"BrokerA"
argument_list|,
name|clientA
argument_list|)
decl_stmt|;
name|MessageIdList
name|msgsB
init|=
name|getConsumerMessages
argument_list|(
literal|"BrokerB"
argument_list|,
name|clientB
argument_list|)
decl_stmt|;
name|MessageIdList
name|msgsC
init|=
name|getConsumerMessages
argument_list|(
literal|"BrokerC"
argument_list|,
name|clientC
argument_list|)
decl_stmt|;
name|msgsA
operator|.
name|waitForMessagesToArrive
argument_list|(
name|MESSAGE_COUNT
operator|*
literal|3
argument_list|)
expr_stmt|;
name|msgsB
operator|.
name|waitForMessagesToArrive
argument_list|(
name|MESSAGE_COUNT
operator|*
literal|3
argument_list|)
expr_stmt|;
name|msgsC
operator|.
name|waitForMessagesToArrive
argument_list|(
name|MESSAGE_COUNT
operator|*
literal|3
argument_list|)
expr_stmt|;
name|assertEquals
argument_list|(
name|MESSAGE_COUNT
operator|*
literal|3
argument_list|,
name|msgsA
operator|.
name|getMessageCount
argument_list|()
argument_list|)
expr_stmt|;
name|assertEquals
argument_list|(
name|MESSAGE_COUNT
operator|*
literal|3
argument_list|,
name|msgsB
operator|.
name|getMessageCount
argument_list|()
argument_list|)
expr_stmt|;
name|assertEquals
argument_list|(
name|MESSAGE_COUNT
operator|*
literal|3
argument_list|,
name|msgsC
operator|.
name|getMessageCount
argument_list|()
argument_list|)
expr_stmt|;
block|}
annotation|@
name|Override
specifier|public
name|void
name|setUp
parameter_list|()
throws|throws
name|Exception
block|{
name|super
operator|.
name|setAutoFail
argument_list|(
literal|true
argument_list|)
expr_stmt|;
name|super
operator|.
name|setUp
argument_list|()
expr_stmt|;
name|String
name|options
init|=
operator|new
name|String
argument_list|(
literal|"?persistent=false&useJmx=false"
argument_list|)
decl_stmt|;
name|createBroker
argument_list|(
operator|new
name|URI
argument_list|(
literal|"broker:(tcp://localhost:61616)/BrokerA"
operator|+
name|options
argument_list|)
argument_list|)
expr_stmt|;
name|createBroker
argument_list|(
operator|new
name|URI
argument_list|(
literal|"broker:(tcp://localhost:61617)/BrokerB"
operator|+
name|options
argument_list|)
argument_list|)
expr_stmt|;
name|createBroker
argument_list|(
operator|new
name|URI
argument_list|(
literal|"broker:(tcp://localhost:61618)/BrokerC"
operator|+
name|options
argument_list|)
argument_list|)
expr_stmt|;
block|}
annotation|@
name|Override
specifier|protected
name|void
name|configureBroker
parameter_list|(
name|BrokerService
name|broker
parameter_list|)
block|{
name|broker
operator|.
name|setBrokerId
argument_list|(
name|broker
operator|.
name|getBrokerName
argument_list|()
argument_list|)
expr_stmt|;
block|}
specifier|public
specifier|static
name|Test
name|suite
parameter_list|()
block|{
return|return
name|suite
argument_list|(
name|ThreeBrokerTopicNetworkTest
operator|.
name|class
argument_list|)
return|;
block|}
block|}
end_class

end_unit

