[hornetq-commits] JBoss hornetq SVN: r9307 - in branches/2_2_0_HA_Improvements: src/config/common/schema and 18 other directories.

Author: timfox Date: 2010-06-11 11:33:36 -0400 (Fri, 11 Jun 2010) New Revision: 9307 Added: branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl_Old.java branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/client/impl/FailoverManagerImpl_Old.java branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/client/impl/FailoverManager_Old.java Removed: branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/client/impl/FailoverManager.java branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/client/impl/FailoverManagerImpl.java Modified: branches/2_2_0_HA_Improvements/build-hornetq.xml branches/2_2_0_HA_Improvements/build.xml branches/2_2_0_HA_Improvements/src/config/common/schema/hornetq-configuration.xsd branches/2_2_0_HA_Improvements/src/main/org/hornetq/api/core/client/ClientSessionFactory.java branches/2_2_0_HA_Improvements/src/main/org/hornetq/api/core/client/HornetQClient.java branches/2_2_0_HA_Improvements/src/main/org/hornetq/api/core/management/HornetQServerControl.java branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/client/impl/ClientSessionFactoryInternal.java branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/client/impl/ClientSessionInternal.java branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/client/impl/DelegatingSession.java branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/cluster/DiscoveryEntry.java branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/cluster/impl/DiscoveryGroupImpl.java branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/config/BroadcastGroupConfiguration.java branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/config/Configuration.java branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/config/impl/ConfigurationImpl.java branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/config/impl/FileConfiguration.java branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/deployers/impl/FileConfigurationParser.java branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/management/impl/BroadcastGroupControlImpl.java branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/management/impl/HornetQServerControlImpl.java branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/persistence/StorageManager.java branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/persistence/impl/nullpm/NullStorageManager.java branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/protocol/core/impl/PacketDecoder.java branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/protocol/core/impl/PacketImpl.java branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/server/cluster/BroadcastGroup.java branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/server/cluster/ClusterManager.java branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/server/cluster/impl/BroadcastGroupImpl.java branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/server/cluster/impl/ClusterManagerImpl.java branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java branches/2_2_0_HA_Improvements/src/main/org/hornetq/utils/UUID.java branches/2_2_0_HA_Improvements/tests/src/org/hornetq/tests/unit/core/paging/impl/PagingStoreImplTest.java Log: new HA Modified: branches/2_2_0_HA_Improvements/build-hornetq.xml =================================================================== --- branches/2_2_0_HA_Improvements/build-hornetq.xml 2010-06-11 04:11:42 UTC (rev 9306) +++ branches/2_2_0_HA_Improvements/build-hornetq.xml 2010-06-11 15:33:36 UTC (rev 9307) @@ -1599,10 +1599,86 @@ <!-- Server execution targets --> <!-- ======================================================================================== --> +<target name="runLive" depends="jar"> + <mkdir dir="logs"/> + <property name="server.config" value="${src.dir}/config/ha-test/live"/> + <java classname="org.hornetq.integration.bootstrap.HornetQBootstrapServer" fork="false"> + <jvmarg value="-XX:+UseParallelGC"/> + <jvmarg value="-Xms512M"/> + <jvmarg value="-Xmx2048M"/> + <jvmarg value="-XX:+AggressiveOpts"/> + <jvmarg value="-XX:+UseFastAccessorMethods"/> + <jvmarg value="-Dcom.sun.management.jmxremote"/> + <jvmarg value="-Djava.util.logging.config.file=${src.config.trunk.non-clustered.dir}/logging.properties"/> + <jvmarg value="-Djava.library.path=${native.bin.dir}"/> + <!--<jvmarg line="-Xdebug -Xrunjdwp:transport=dt_socket,server=y,suspend=y,address=8000"/>--> + <arg line="hornetq-beans.xml"/> + <classpath path="${server.config}" /> + <classpath refid="jms.standalone.server.classpath"/> + </java> + </target> + +<target name="runBackup1" depends="jar"> + <mkdir dir="logs"/> + <property name="server.config" value="${src.dir}/config/ha-test/backup-1"/> + <java classname="org.hornetq.integration.bootstrap.HornetQBootstrapServer" fork="false"> + <jvmarg value="-XX:+UseParallelGC"/> + <jvmarg value="-Xms512M"/> + <jvmarg value="-Xmx2048M"/> + <jvmarg value="-XX:+AggressiveOpts"/> + <jvmarg value="-XX:+UseFastAccessorMethods"/> + <jvmarg value="-Dcom.sun.management.jmxremote"/> + <jvmarg value="-Djava.util.logging.config.file=${src.config.trunk.non-clustered.dir}/logging.properties"/> + <jvmarg value="-Djava.library.path=${native.bin.dir}"/> + <!--<jvmarg line="-Xdebug -Xrunjdwp:transport=dt_socket,server=y,suspend=y,address=8000"/>--> + <arg line="hornetq-beans.xml"/> + <classpath path="${server.config}" /> + <classpath refid="jms.standalone.server.classpath"/> + </java> + </target> + +<target name="runBackup2" depends="jar"> + <mkdir dir="logs"/> + <property name="server.config" value="${src.dir}/config/ha-test/backup-2"/> + <java classname="org.hornetq.integration.bootstrap.HornetQBootstrapServer" fork="false"> + <jvmarg value="-XX:+UseParallelGC"/> + <jvmarg value="-Xms512M"/> + <jvmarg value="-Xmx2048M"/> + <jvmarg value="-XX:+AggressiveOpts"/> + <jvmarg value="-XX:+UseFastAccessorMethods"/> + <jvmarg value="-Dcom.sun.management.jmxremote"/> + <jvmarg value="-Djava.util.logging.config.file=${src.config.trunk.non-clustered.dir}/logging.properties"/> + <jvmarg value="-Djava.library.path=${native.bin.dir}"/> + <!--<jvmarg line="-Xdebug -Xrunjdwp:transport=dt_socket,server=y,suspend=y,address=8000"/>--> + <arg line="hornetq-beans.xml"/> + <classpath path="${server.config}" /> + <classpath refid="jms.standalone.server.classpath"/> + </java> + </target> + +<target name="runBackup3" depends="jar"> + <mkdir dir="logs"/> + <property name="server.config" value="${src.dir}/config/ha-test/backup-3"/> + <java classname="org.hornetq.integration.bootstrap.HornetQBootstrapServer" fork="false"> + <jvmarg value="-XX:+UseParallelGC"/> + <jvmarg value="-Xms512M"/> + <jvmarg value="-Xmx2048M"/> + <jvmarg value="-XX:+AggressiveOpts"/> + <jvmarg value="-XX:+UseFastAccessorMethods"/> + <jvmarg value="-Dcom.sun.management.jmxremote"/> + <jvmarg value="-Djava.util.logging.config.file=${src.config.trunk.non-clustered.dir}/logging.properties"/> + <jvmarg value="-Djava.library.path=${native.bin.dir}"/> + <!--<jvmarg line="-Xdebug -Xrunjdwp:transport=dt_socket,server=y,suspend=y,address=8000"/>--> + <arg line="hornetq-beans.xml"/> + <classpath path="${server.config}" /> + <classpath refid="jms.standalone.server.classpath"/> + </java> + </target> + <target name="runServer" depends="jar"> <mkdir dir="logs"/> <property name="server.config" value="${src.config.trunk.non-clustered.dir}"/> - <java classname="org.hornetq.integration.bootstrap.HornetQBootstrapServer" fork="true"> + <java classname="org.hornetq.integration.bootstrap.HornetQBootstrapServer" fork="false"> <jvmarg value="-XX:+UseParallelGC"/> <jvmarg value="-Xms512M"/> <jvmarg value="-Xmx2048M"/> Modified: branches/2_2_0_HA_Improvements/build.xml =================================================================== --- branches/2_2_0_HA_Improvements/build.xml 2010-06-11 04:11:42 UTC (rev 9306) +++ branches/2_2_0_HA_Improvements/build.xml 2010-06-11 15:33:36 UTC (rev 9307) @@ -319,6 +319,22 @@ <ant antfile="build-hornetq.xml" target="runServer"/> </target> + <target name="runLive" depends="createthirdparty"> + <ant antfile="build-hornetq.xml" target="runLive"/> + </target> + + <target name="runBackup1" depends="createthirdparty"> + <ant antfile="build-hornetq.xml" target="runBackup1"/> + </target> + + <target name="runBackup2" depends="createthirdparty"> + <ant antfile="build-hornetq.xml" target="runBackup2"/> + </target> + + <target name="runBackup3" depends="createthirdparty"> + <ant antfile="build-hornetq.xml" target="runBackup3"/> + </target> + <target name="runClusteredServer" depends="createthirdparty"> <ant antfile="build-hornetq.xml" target="runClusteredServer"/> </target> Modified: branches/2_2_0_HA_Improvements/src/config/common/schema/hornetq-configuration.xsd =================================================================== --- branches/2_2_0_HA_Improvements/src/config/common/schema/hornetq-configuration.xsd 2010-06-11 04:11:42 UTC (rev 9306) +++ branches/2_2_0_HA_Improvements/src/config/common/schema/hornetq-configuration.xsd 2010-06-11 15:33:36 UTC (rev 9307) @@ -71,7 +71,7 @@ </xsd:element> <xsd:element maxOccurs="1" minOccurs="0" name="persist-delivery-count-before-delivery" type="xsd:boolean"> </xsd:element> - <xsd:element maxOccurs="1" minOccurs="0" name="backup-connector-ref" type="backup-connectorType"> + <xsd:element maxOccurs="1" minOccurs="0" name="live-connector-ref" type="live-connectorType"> </xsd:element> <xsd:element maxOccurs="1" minOccurs="0" name="connectors"> <xsd:complexType> @@ -492,7 +492,7 @@ </xsd:complexType> </xsd:element> - <xsd:complexType name="backup-connectorType"> + <xsd:complexType name="live-connectorType"> <xsd:attribute name="connector-name" type="xsd:IDREF" use="required"> </xsd:attribute> </xsd:complexType> Modified: branches/2_2_0_HA_Improvements/src/main/org/hornetq/api/core/client/ClientSessionFactory.java =================================================================== --- branches/2_2_0_HA_Improvements/src/main/org/hornetq/api/core/client/ClientSessionFactory.java 2010-06-11 04:11:42 UTC (rev 9306) +++ branches/2_2_0_HA_Improvements/src/main/org/hornetq/api/core/client/ClientSessionFactory.java 2010-06-11 15:33:36 UTC (rev 9307) @@ -13,14 +13,9 @@ package org.hornetq.api.core.client; -import java.util.List; - import org.hornetq.api.core.HornetQException; -import org.hornetq.api.core.Interceptor; -import org.hornetq.api.core.Pair; -import org.hornetq.api.core.TransportConfiguration; -import org.hornetq.api.core.client.loadbalance.ConnectionLoadBalancingPolicy; + /** * A ClientSessionFactory is the entry point to create and configure HornetQ resources to produce and consume messages. * <br> @@ -136,654 +131,634 @@ boolean preAcknowledge, int ackBatchSize) throws HornetQException; - /** - * Returns the list of <em>live - backup</em> connectors pairs configured - * that sessions created by this factory will use to connect - * to HornetQ servers or <code>null</code> if the factory is using discovery group. - * - * The backup configuration (returned by {@link org.hornetq.api.core.Pair#b}) can be <code>null</code> if there is no - * backup for the corresponding live configuration (returned by {@link org.hornetq.api.core.Pair#a}) - * - * @return a list of pair of TransportConfiguration corresponding to the live - backup nodes - */ - List<Pair<TransportConfiguration, TransportConfiguration>> getStaticConnectors(); - - /** - * Sets the static list of live - backup connectors pairs that sessions created by this factory will use to connect - * to HornetQ servers. - * - * The backup configuration (returned by {@link Pair#b}) can be <code>null</code> if there is no - * backup for the corresponding live configuration (returned by {@link Pair#a}) - * - * @param staticConnectors a list of pair of TransportConfiguration corresponding to the live - backup nodes - */ - void setStaticConnectors(List<Pair<TransportConfiguration, TransportConfiguration>> staticConnectors); - - /** - * Returns the period used to check if a client has failed to receive pings from the server. - * - * Period is in milliseconds, default value is {@link HornetQClient#DEFAULT_CLIENT_FAILURE_CHECK_PERIOD}. - * - * @return the period used to check if a client has failed to receive pings from the server - */ - long getClientFailureCheckPeriod(); - - /** - * Sets the period (in milliseconds) used to check if a client has failed to receive pings from the server. - * - * Value must be -1 (to disable) or greater than 0. - * - * @param clientFailureCheckPeriod the period to check failure - */ - void setClientFailureCheckPeriod(long clientFailureCheckPeriod); - - /** - * When <code>true</code>, consumers created through this factory will create temporary files to cache large messages. - * - * There is 1 temporary file created for each large message. - * - * Default value is {@link HornetQClient#DEFAULT_CACHE_LARGE_MESSAGE_CLIENT}. - * - * @return <code>true</code> if consumers created through this factory will cache large messages in temporary files, <code>false</code> else - */ - boolean isCacheLargeMessagesClient(); - - /** - * Sets whether large messages received by consumers created through this factory will be cached in temporary files or not. - * - * @param cached <code>true</code> to cache large messages in temporary files, <code>false</code> else - */ - void setCacheLargeMessagesClient(boolean cached); - - /** - * Returns the connection <em>time-to-live</em>. - * This TTL determines how long the server will keep a connection alive in the absence of any data arriving from the client. - * - * Value is in milliseconds, default value is {@link HornetQClient#DEFAULT_CONNECTION_TTL}. - * - * @return the connection time-to-live in milliseconds - */ - long getConnectionTTL(); - - /** - * Sets this factory's connections <em>time-to-live</em>. - * - * Value must be -1 (to disable) or greater or equals to 0. - * - * @param connectionTTL period in milliseconds - */ - void setConnectionTTL(long connectionTTL); - - /** - * Returns the blocking calls timeout. - * - * If client's blocking calls to the server take more than this timeout, the call will throw a {@link HornetQException} with the code {@link HornetQException#CONNECTION_TIMEDOUT}. - * Value is in milliseconds, default value is {@link HornetQClient#DEFAULT_CALL_TIMEOUT}. - * - * @return the blocking calls timeout - */ - long getCallTimeout(); - - /** - * Sets the blocking call timeout. - * - * Value must be greater or equals to 0 - * - * @param callTimeout blocking call timeout in milliseconds - */ - void setCallTimeout(long callTimeout); - - /** - * Returns the large message size threshold. - * - * Messages whose size is if greater than this value will be handled as <em>large messages</em>. - * - * Value is in bytes, default value is {@link HornetQClient#DEFAULT_MIN_LARGE_MESSAGE_SIZE}. - * - * @return the message size threshold to treat messages as large messages. - */ - int getMinLargeMessageSize(); - - /** - * Sets the large message size threshold. - * - * Value must be greater than 0. - * - * @param minLargeMessageSize large message size threshold in bytes - */ - void setMinLargeMessageSize(int minLargeMessageSize); - - /** - * Returns the window size for flow control of the consumers created through this factory. - * - * Value is in bytes, default value is {@link HornetQClient#DEFAULT_CONSUMER_WINDOW_SIZE}. - * - * @return the window size used for consumer flow control - */ - int getConsumerWindowSize(); - - /** - * Sets the window size for flow control of the consumers created through this factory. - * - * Value must be -1 (to disable flow control), 0 (to not buffer any messages) or greater than 0 (to set the maximum size of the buffer) - * - * @param consumerWindowSize window size (in bytes) used for consumer flow control - */ - void setConsumerWindowSize(int consumerWindowSize); - - /** - * Returns the maximum rate of message consumption for consumers created through this factory. - * - * This value controls the rate at which a consumer can consume messages. A consumer will never consume messages at a rate faster than the rate specified. - * - * Value is -1 (to disable) or a positive integer corresponding to the maximum desired message consumption rate specified in units of messages per second. - * Default value is {@link HornetQClient#DEFAULT_CONSUMER_MAX_RATE}. - * - * @return the consumer max rate - */ - int getConsumerMaxRate(); - - /** - * Sets the maximum rate of message consumption for consumers created through this factory. - * - * Value must -1 (to disable) or a positive integer corresponding to the maximum desired message consumption rate specified in units of messages per second. - * - * @param consumerMaxRate maximum rate of message consumption (in messages per seconds) - */ - void setConsumerMaxRate(int consumerMaxRate); - - /** - * Returns the size for the confirmation window of clients using this factory. - * - * Value is in bytes or -1 (to disable the window). Default value is {@link HornetQClient#DEFAULT_CONFIRMATION_WINDOW_SIZE}. - * - * @return the size for the confirmation window of clients using this factory - */ - int getConfirmationWindowSize(); - - /** - * Sets the size for the confirmation window buffer of clients using this factory. - * - * Value must be -1 (to disable the window) or greater than 0. - - * @param confirmationWindowSize size of the confirmation window (in bytes) - */ - void setConfirmationWindowSize(int confirmationWindowSize); - - /** - * Returns the window size for flow control of the producers created through this factory. - * - * Value must be -1 (to disable flow control) or greater than 0 to determine the maximum amount of bytes at any give time (to prevent overloading the connection). - * Default value is {@link HornetQClient#DEFAULT_PRODUCER_WINDOW_SIZE}. - * - * @return the window size for flow control of the producers created through this factory. - */ - int getProducerWindowSize(); - - /** - * Returns the window size for flow control of the producers created through this factory. - * - * Value must be -1 (to disable flow control) or greater than 0. - * - * @param producerWindowSize window size (in bytest) for flow control of the producers created through this factory. - */ - void setProducerWindowSize(int producerWindowSize); - - /** - * Returns the maximum rate of message production for producers created through this factory. - * - * This value controls the rate at which a producer can produce messages. A producer will never produce messages at a rate faster than the rate specified. - * - * Value is -1 (to disable) or a positive integer corresponding to the maximum desired message production rate specified in units of messages per second. - * Default value is {@link HornetQClient#DEFAULT_PRODUCER_MAX_RATE}. - * - * @return maximum rate of message production (in messages per seconds) - */ - int getProducerMaxRate(); - - /** - * Sets the maximum rate of message production for producers created through this factory. - * - * Value must -1 (to disable) or a positive integer corresponding to the maximum desired message production rate specified in units of messages per second. - * - * @param producerMaxRate maximum rate of message production (in messages per seconds) - */ - void setProducerMaxRate(int producerMaxRate); - - /** - * Returns whether consumers created through this factory will block while sending message acknowledgements or do it asynchronously. - * - * Default value is {@link HornetQClient#DEFAULT_BLOCK_ON_ACKNOWLEDGE}. - * - * @return whether consumers will block while sending message acknowledgements or do it asynchronously - */ - boolean isBlockOnAcknowledge(); - - /** - * Sets whether consumers created through this factory will block while sending message acknowledgements or do it asynchronously. - * - * @param blockOnAcknowledge <code>true</code> to block when sending message acknowledgements or <code>false</code> to send them asynchronously - */ - void setBlockOnAcknowledge(boolean blockOnAcknowledge); - - /** - * Returns whether producers created through this factory will block while sending <em>durable</em> messages or do it asynchronously. - * <br> - * If the session is configured to send durable message asynchronously, the client can set a SendAcknowledgementHandler on the ClientSession - * to be notified once the message has been handled by the server. - * - * Default value is {@link HornetQClient#DEFAULT_BLOCK_ON_DURABLE_SEND}. - * - * @return whether producers will block while sending persistent messages or do it asynchronously - */ - boolean isBlockOnDurableSend(); - - /** - * Sets whether producers created through this factory will block while sending <em>durable</em> messages or do it asynchronously. - * - * @param blockOnDurableSend <code>true</code> to block when sending durable messages or <code>false</code> to send them asynchronously - */ - void setBlockOnDurableSend(boolean blockOnDurableSend); - - /** - * Returns whether producers created through this factory will block while sending <em>non-durable</em> messages or do it asynchronously. - * <br> - * If the session is configured to send non-durable message asynchronously, the client can set a SendAcknowledgementHandler on the ClientSession - * to be notified once the message has been handled by the server. - * - * Default value is {@link HornetQClient#DEFAULT_BLOCK_ON_NON_DURABLE_SEND}. - * - * @return whether producers will block while sending non-durable messages or do it asynchronously - */ - boolean isBlockOnNonDurableSend(); - - /** - * Sets whether producers created through this factory will block while sending <em>non-durable</em> messages or do it asynchronously. - * - * @param blockOnNonDurableSend <code>true</code> to block when sending non-durable messages or <code>false</code> to send them asynchronously - */ - void setBlockOnNonDurableSend(boolean blockOnNonDurableSend); - - /** - * Returns whether producers created through this factory will automatically - * assign a group ID to the messages they sent. - * - * if <code>true</code>, a random unique group ID is created and set on each message for the property - * {@link org.hornetq.api.core.Message#HDR_GROUP_ID}. - * Default value is {@link HornetQClient#DEFAULT_AUTO_GROUP}. - * - * @return whether producers will automatically assign a group ID to their messages - */ - boolean isAutoGroup(); - - /** - * Sets whether producers created through this factory will automatically - * assign a group ID to the messages they sent. - * - * @param autoGroup <code>true</code> to automatically assign a group ID to each messages sent through this factory, <code>false</code> else - */ - void setAutoGroup(boolean autoGroup); - - /** - * Returns the group ID that will be eventually set on each message for the property {@link org.hornetq.api.core.Message#HDR_GROUP_ID}. - * - * Default value is is <code>null</code> and no group ID will be set on the messages. - * - * @return the group ID that will be eventually set on each message - */ - String getGroupID(); - /** - * Sets the group ID that will be set on each message sent through this factory. - * - * @param groupID the group ID to use - */ - void setGroupID(String groupID); +// /** +// * Returns the period used to check if a client has failed to receive pings from the server. +// * +// * Period is in milliseconds, default value is {@link HornetQClient#DEFAULT_CLIENT_FAILURE_CHECK_PERIOD}. +// * +// * @return the period used to check if a client has failed to receive pings from the server +// */ +// long getClientFailureCheckPeriod(); +// +// /** +// * Sets the period (in milliseconds) used to check if a client has failed to receive pings from the server. +// * +// * Value must be -1 (to disable) or greater than 0. +// * +// * @param clientFailureCheckPeriod the period to check failure +// */ +// void setClientFailureCheckPeriod(long clientFailureCheckPeriod); +// +// /** +// * When <code>true</code>, consumers created through this factory will create temporary files to cache large messages. +// * +// * There is 1 temporary file created for each large message. +// * +// * Default value is {@link HornetQClient#DEFAULT_CACHE_LARGE_MESSAGE_CLIENT}. +// * +// * @return <code>true</code> if consumers created through this factory will cache large messages in temporary files, <code>false</code> else +// */ +// boolean isCacheLargeMessagesClient(); +// +// /** +// * Sets whether large messages received by consumers created through this factory will be cached in temporary files or not. +// * +// * @param cached <code>true</code> to cache large messages in temporary files, <code>false</code> else +// */ +// void setCacheLargeMessagesClient(boolean cached); +// +// /** +// * Returns the connection <em>time-to-live</em>. +// * This TTL determines how long the server will keep a connection alive in the absence of any data arriving from the client. +// * +// * Value is in milliseconds, default value is {@link HornetQClient#DEFAULT_CONNECTION_TTL}. +// * +// * @return the connection time-to-live in milliseconds +// */ +// long getConnectionTTL(); +// +// /** +// * Sets this factory's connections <em>time-to-live</em>. +// * +// * Value must be -1 (to disable) or greater or equals to 0. +// * +// * @param connectionTTL period in milliseconds +// */ +// void setConnectionTTL(long connectionTTL); +// +// /** +// * Returns the blocking calls timeout. +// * +// * If client's blocking calls to the server take more than this timeout, the call will throw a {@link HornetQException} with the code {@link HornetQException#CONNECTION_TIMEDOUT}. +// * Value is in milliseconds, default value is {@link HornetQClient#DEFAULT_CALL_TIMEOUT}. +// * +// * @return the blocking calls timeout +// */ +// long getCallTimeout(); +// +// /** +// * Sets the blocking call timeout. +// * +// * Value must be greater or equals to 0 +// * +// * @param callTimeout blocking call timeout in milliseconds +// */ +// void setCallTimeout(long callTimeout); +// +// /** +// * Returns the large message size threshold. +// * +// * Messages whose size is if greater than this value will be handled as <em>large messages</em>. +// * +// * Value is in bytes, default value is {@link HornetQClient#DEFAULT_MIN_LARGE_MESSAGE_SIZE}. +// * +// * @return the message size threshold to treat messages as large messages. +// */ +// int getMinLargeMessageSize(); +// +// /** +// * Sets the large message size threshold. +// * +// * Value must be greater than 0. +// * +// * @param minLargeMessageSize large message size threshold in bytes +// */ +// void setMinLargeMessageSize(int minLargeMessageSize); +// +// /** +// * Returns the window size for flow control of the consumers created through this factory. +// * +// * Value is in bytes, default value is {@link HornetQClient#DEFAULT_CONSUMER_WINDOW_SIZE}. +// * +// * @return the window size used for consumer flow control +// */ +// int getConsumerWindowSize(); +// +// /** +// * Sets the window size for flow control of the consumers created through this factory. +// * +// * Value must be -1 (to disable flow control), 0 (to not buffer any messages) or greater than 0 (to set the maximum size of the buffer) +// * +// * @param consumerWindowSize window size (in bytes) used for consumer flow control +// */ +// void setConsumerWindowSize(int consumerWindowSize); +// +// /** +// * Returns the maximum rate of message consumption for consumers created through this factory. +// * +// * This value controls the rate at which a consumer can consume messages. A consumer will never consume messages at a rate faster than the rate specified. +// * +// * Value is -1 (to disable) or a positive integer corresponding to the maximum desired message consumption rate specified in units of messages per second. +// * Default value is {@link HornetQClient#DEFAULT_CONSUMER_MAX_RATE}. +// * +// * @return the consumer max rate +// */ +// int getConsumerMaxRate(); +// +// /** +// * Sets the maximum rate of message consumption for consumers created through this factory. +// * +// * Value must -1 (to disable) or a positive integer corresponding to the maximum desired message consumption rate specified in units of messages per second. +// * +// * @param consumerMaxRate maximum rate of message consumption (in messages per seconds) +// */ +// void setConsumerMaxRate(int consumerMaxRate); +// +// /** +// * Returns the size for the confirmation window of clients using this factory. +// * +// * Value is in bytes or -1 (to disable the window). Default value is {@link HornetQClient#DEFAULT_CONFIRMATION_WINDOW_SIZE}. +// * +// * @return the size for the confirmation window of clients using this factory +// */ +// int getConfirmationWindowSize(); +// +// /** +// * Sets the size for the confirmation window buffer of clients using this factory. +// * +// * Value must be -1 (to disable the window) or greater than 0. +// +// * @param confirmationWindowSize size of the confirmation window (in bytes) +// */ +// void setConfirmationWindowSize(int confirmationWindowSize); +// +// /** +// * Returns the window size for flow control of the producers created through this factory. +// * +// * Value must be -1 (to disable flow control) or greater than 0 to determine the maximum amount of bytes at any give time (to prevent overloading the connection). +// * Default value is {@link HornetQClient#DEFAULT_PRODUCER_WINDOW_SIZE}. +// * +// * @return the window size for flow control of the producers created through this factory. +// */ +// int getProducerWindowSize(); +// +// /** +// * Returns the window size for flow control of the producers created through this factory. +// * +// * Value must be -1 (to disable flow control) or greater than 0. +// * +// * @param producerWindowSize window size (in bytest) for flow control of the producers created through this factory. +// */ +// void setProducerWindowSize(int producerWindowSize); +// +// /** +// * Returns the maximum rate of message production for producers created through this factory. +// * +// * This value controls the rate at which a producer can produce messages. A producer will never produce messages at a rate faster than the rate specified. +// * +// * Value is -1 (to disable) or a positive integer corresponding to the maximum desired message production rate specified in units of messages per second. +// * Default value is {@link HornetQClient#DEFAULT_PRODUCER_MAX_RATE}. +// * +// * @return maximum rate of message production (in messages per seconds) +// */ +// int getProducerMaxRate(); +// +// /** +// * Sets the maximum rate of message production for producers created through this factory. +// * +// * Value must -1 (to disable) or a positive integer corresponding to the maximum desired message production rate specified in units of messages per second. +// * +// * @param producerMaxRate maximum rate of message production (in messages per seconds) +// */ +// void setProducerMaxRate(int producerMaxRate); +// +// /** +// * Returns whether consumers created through this factory will block while sending message acknowledgements or do it asynchronously. +// * +// * Default value is {@link HornetQClient#DEFAULT_BLOCK_ON_ACKNOWLEDGE}. +// * +// * @return whether consumers will block while sending message acknowledgements or do it asynchronously +// */ +// boolean isBlockOnAcknowledge(); +// +// /** +// * Sets whether consumers created through this factory will block while sending message acknowledgements or do it asynchronously. +// * +// * @param blockOnAcknowledge <code>true</code> to block when sending message acknowledgements or <code>false</code> to send them asynchronously +// */ +// void setBlockOnAcknowledge(boolean blockOnAcknowledge); +// +// /** +// * Returns whether producers created through this factory will block while sending <em>durable</em> messages or do it asynchronously. +// * <br> +// * If the session is configured to send durable message asynchronously, the client can set a SendAcknowledgementHandler on the ClientSession +// * to be notified once the message has been handled by the server. +// * +// * Default value is {@link HornetQClient#DEFAULT_BLOCK_ON_DURABLE_SEND}. +// * +// * @return whether producers will block while sending persistent messages or do it asynchronously +// */ +// boolean isBlockOnDurableSend(); +// +// /** +// * Sets whether producers created through this factory will block while sending <em>durable</em> messages or do it asynchronously. +// * +// * @param blockOnDurableSend <code>true</code> to block when sending durable messages or <code>false</code> to send them asynchronously +// */ +// void setBlockOnDurableSend(boolean blockOnDurableSend); +// +// /** +// * Returns whether producers created through this factory will block while sending <em>non-durable</em> messages or do it asynchronously. +// * <br> +// * If the session is configured to send non-durable message asynchronously, the client can set a SendAcknowledgementHandler on the ClientSession +// * to be notified once the message has been handled by the server. +// * +// * Default value is {@link HornetQClient#DEFAULT_BLOCK_ON_NON_DURABLE_SEND}. +// * +// * @return whether producers will block while sending non-durable messages or do it asynchronously +// */ +// boolean isBlockOnNonDurableSend(); +// +// /** +// * Sets whether producers created through this factory will block while sending <em>non-durable</em> messages or do it asynchronously. +// * +// * @param blockOnNonDurableSend <code>true</code> to block when sending non-durable messages or <code>false</code> to send them asynchronously +// */ +// void setBlockOnNonDurableSend(boolean blockOnNonDurableSend); +// +// /** +// * Returns whether producers created through this factory will automatically +// * assign a group ID to the messages they sent. +// * +// * if <code>true</code>, a random unique group ID is created and set on each message for the property +// * {@link org.hornetq.api.core.Message#HDR_GROUP_ID}. +// * Default value is {@link HornetQClient#DEFAULT_AUTO_GROUP}. +// * +// * @return whether producers will automatically assign a group ID to their messages +// */ +// boolean isAutoGroup(); +// +// /** +// * Sets whether producers created through this factory will automatically +// * assign a group ID to the messages they sent. +// * +// * @param autoGroup <code>true</code> to automatically assign a group ID to each messages sent through this factory, <code>false</code> else +// */ +// void setAutoGroup(boolean autoGroup); +// +// /** +// * Returns the group ID that will be eventually set on each message for the property {@link org.hornetq.api.core.Message#HDR_GROUP_ID}. +// * +// * Default value is is <code>null</code> and no group ID will be set on the messages. +// * +// * @return the group ID that will be eventually set on each message +// */ +// String getGroupID(); +// +// /** +// * Sets the group ID that will be set on each message sent through this factory. +// * +// * @param groupID the group ID to use +// */ +// void setGroupID(String groupID); +// +// /** +// * Returns whether messages will pre-acknowledged on the server before they are sent to the consumers or not. +// * +// * Default value is {@link HornetQClient#DEFAULT_PRE_ACKNOWLEDGE} +// */ +// boolean isPreAcknowledge(); +// +// /** +// * Sets to <code>true</code> to pre-acknowledge consumed messages on the server before they are sent to consumers, else set to <code>false</code> to let +// * clients acknowledge the message they consume. +// * +// * @param preAcknowledge <code>true</code> to enable pre-acknowledgement, <code>false</code> else +// */ +// void setPreAcknowledge(boolean preAcknowledge); +// +// /** +// * Returns the acknowledgements batch size. +// * +// * Default value is {@link HornetQClient#DEFAULT_ACK_BATCH_SIZE}. +// * +// * @return the acknowledgements batch size +// */ +// int getAckBatchSize(); +// +// /** +// * Sets the acknowledgements batch size. +// * +// * Value must be equal or greater than 0. +// * +// * @param ackBatchSize acknowledgements batch size +// */ +// void setAckBatchSize(int ackBatchSize); +// +// /** +// * Returns the local bind address to which the multicast socket is bound for discovery. +// * +// * This is null if the multicast socket is not bound, or no discovery is being used +// * +// * @return the local bind address to which the multicast socket is bound for discovery +// */ +// String getLocalBindAddress(); +// +// /** +// * Sets the local bind address to which the multicast socket is bound for discovery. +// * +// * @param the local bind address +// */ +// void setLocalBindAddress(String localBindAddress); +// +// /** +// * Returns the address to listen to discover which connectors this factory can use. +// * The discovery address must be set to enable this factory to discover HornetQ servers. +// * +// * @return the address to listen to discover which connectors this factory can use +// */ +// String getDiscoveryAddress(); +// +// /** +// * Sets the address to listen to discover which connectors this factory can use. +// * +// * @param discoveryAddress address to listen to discover which connectors this factory can use +// */ +// void setDiscoveryAddress(String discoveryAddress); +// +// /** +// * Returns the port to listen to discover which connectors this factory can use. +// * The discovery port must be set to enable this factory to discover HornetQ servers. +// * +// * @return the port to listen to discover which connectors this factory can use +// */ +// int getDiscoveryPort(); +// +// +// /** +// * Sets the port to listen to discover which connectors this factory can use. +// * +// * @param discoveryPort port to listen to discover which connectors this factory can use +// */ +// void setDiscoveryPort(int discoveryPort); +// +// /** +// * Returns the refresh timeout for discovered HornetQ servers. +// * +// * If this factory uses discovery to find HornetQ servers, the list of discovered servers +// * will be refreshed according to this timeout. +// * +// * Value is in milliseconds, default value is {@link HornetQClient#DEFAULT_DISCOVERY_REFRESH_TIMEOUT}. +// * +// * @return the refresh timeout for discovered HornetQ servers +// */ +// long getDiscoveryRefreshTimeout(); +// +// /** +// * Sets the refresh timeout for discovered HornetQ servers. +// * +// * Value must be greater than 0. +// * +// * @param discoveryRefreshTimeout refresh timeout (in milliseconds) for discovered HornetQ servers +// */ +// void setDiscoveryRefreshTimeout(long discoveryRefreshTimeout); +// +// /** +// * Returns the initial wait timeout if this factory is configured to use discovery. +// * +// * Value is in milliseconds, default value is {@link HornetQClient#DEFAULT_DISCOVERY_INITIAL_WAIT_TIMEOUT}. +// * +// * @return the initial wait timeout if this factory is configured to use discovery +// */ +// long getDiscoveryInitialWaitTimeout(); +// +// /** +// * Sets the initial wait timeout if this factory is configured to use discovery. +// * +// * Value is in milliseconds and must be greater than 0. +// * +// * @param initialWaitTimeout initial wait timeout when using discovery +// */ +// void setDiscoveryInitialWaitTimeout(long initialWaitTimeout); +// +// /** +// * Returns whether this factory will use global thread pools (shared among all the factories in the same JVM) +// * or its own pools. +// * +// * Default value is {@link HornetQClient#DEFAULT_USE_GLOBAL_POOLS}. +// * +// * @return <code>true</code> if this factory uses global thread pools, <code>false</code> else +// */ +// boolean isUseGlobalPools(); +// +// /** +// * Sets whether this factory will use global thread pools (shared among all the factories in the same JVM) +// * or its own pools. +// * +// * @param useGlobalPools <code>true</code> to let this factory uses global thread pools, <code>false</code> else +// */ +// void setUseGlobalPools(boolean useGlobalPools); +// +// /** +// * Returns the maximum size of the scheduled thread pool. +// * +// * Default value is {@link HornetQClient#DEFAULT_SCHEDULED_THREAD_POOL_MAX_SIZE}. +// * +// * @return the maximum size of the scheduled thread pool. +// */ +// int getScheduledThreadPoolMaxSize(); +// +// /** +// * Sets the maximum size of the scheduled thread pool. +// * +// * This setting is relevant only if this factory does not use global pools. +// * Value must be greater than 0. +// * +// * @param scheduledThreadPoolMaxSize maximum size of the scheduled thread pool. +// */ +// void setScheduledThreadPoolMaxSize(int scheduledThreadPoolMaxSize); +// +// /** +// * Returns the maximum size of the thread pool. +// * +// * Default value is {@link HornetQClient#DEFAULT_THREAD_POOL_MAX_SIZE}. +// * +// * @return the maximum size of the thread pool. +// */ +// int getThreadPoolMaxSize(); +// +// /** +// * Sets the maximum size of the thread pool. +// * +// * This setting is relevant only if this factory does not use global pools. +// * Value must be -1 (for unlimited thread pool) or greater than 0. +// * +// * @param threadPoolMaxSize maximum size of the thread pool. +// */ +// void setThreadPoolMaxSize(int threadPoolMaxSize); +// +// /** +// * Returns the time to retry connections created by this factory after failure. +// * +// * Value is in milliseconds, default is {@link HornetQClient#DEFAULT_RETRY_INTERVAL}. +// * +// * @return the time to retry connections created by this factory after failure +// */ +// long getRetryInterval(); +// +// /** +// * Sets the time to retry connections created by this factory after failure. +// * +// * Value must be greater than 0. +// * +// * @param retryInterval time (in milliseconds) to retry connections created by this factory after failure +// */ +// void setRetryInterval(long retryInterval); +// +// /** +// * Returns the multiplier to apply to successive retry intervals. +// * +// * Default value is {@link HornetQClient#DEFAULT_RETRY_INTERVAL_MULTIPLIER}. +// * +// * @return the multiplier to apply to successive retry intervals +// */ +// double getRetryIntervalMultiplier(); +// +// /** +// * Sets the multiplier to apply to successive retry intervals. +// * +// * Value must be positive. +// * +// * @param retryIntervalMultiplier multiplier to apply to successive retry intervals +// */ +// void setRetryIntervalMultiplier(double retryIntervalMultiplier); +// +// /** +// * Returns the maximum retry interval (in the case a retry interval multiplier has been specified). +// * +// * Value is in milliseconds, default value is {@link HornetQClient#DEFAULT_MAX_RETRY_INTERVAL}. +// * +// * @return the maximum retry interval +// */ +// long getMaxRetryInterval(); +// +// /** +// * Sets the maximum retry interval. +// * +// * Value must be greater than 0. +// * +// * @param maxRetryInterval maximum retry interval to apply in the case a retry interval multiplier has been specified +// */ +// void setMaxRetryInterval(long maxRetryInterval); +// +// /** +// * Returns the maximum number of attempts to retry connection in case of failure. +// * +// * Default value is {@link HornetQClient#DEFAULT_RECONNECT_ATTEMPTS}. +// * +// * @return the maximum number of attempts to retry connection in case of failure. +// */ +// int getReconnectAttempts(); +// +// /** +// * Sets the maximum number of attempts to retry connection in case of failure. +// * +// * Value must be -1 (to retry infinitely), 0 (to never retry connection) or greater than 0. +// * +// * @param reconnectAttempts maximum number of attempts to retry connection in case of failure +// */ +// void setReconnectAttempts(int reconnectAttempts); +// +// /** +// * Returns true if the client will automatically attempt to connect to the backup server if the initial +// * connection to the live server fails +// * +// * Default value is {@link HornetQClient.DEFAULT_FAILOVER_ON_INITIAL_CONNECTION}. +// * +// * @return +// */ +// boolean isFailoverOnInitialConnection(); +// +// /** +// * Sets the value for FailoverOnInitialReconnection +// * +// * @param failover +// */ +// void setFailoverOnInitialConnection(boolean failover); +// +// /** +// * Returns whether connections created by this factory must failover in case the server they are +// * connected to <em>has normally shut down</em>. +// * +// * Default value is {@link HornetQClient#DEFAULT_FAILOVER_ON_SERVER_SHUTDOWN}. +// * +// * @return <code>true</code> if connections must failover if the server has normally shut down, else <code>false</code> +// */ +// boolean isFailoverOnServerShutdown(); +// +// /** +// * Sets whether connections created by this factory must failover in case the server they are +// * connected to <em>has normally shut down</em> +// * +// * @param failoverOnServerShutdown <code>true</code> if connections must failover if the server has normally shut down, <code>false</code> else +// */ +// void setFailoverOnServerShutdown(boolean failoverOnServerShutdown); +// +// /** +// * Returns the class name of the connection load balancing policy. +// * +// * Default value is "org.hornetq.api.core.client.loadbalance.RoundRobinConnectionLoadBalancingPolicy". +// * +// * @return the class name of the connection load balancing policy +// */ +// String getConnectionLoadBalancingPolicyClassName(); +// +// /** +// * Sets the class name of the connection load balancing policy. +// * +// * Value must be the name of a class implementing {@link ConnectionLoadBalancingPolicy}. +// * +// * @param loadBalancingPolicyClassName class name of the connection load balancing policy +// */ +// void setConnectionLoadBalancingPolicyClassName(String loadBalancingPolicyClassName); +// +// /** +// * Returns the initial size of messages created through this factory. +// * +// * Value is in bytes, default value is {@link HornetQClient#DEFAULT_INITIAL_MESSAGE_PACKET_SIZE}. +// * +// * @return the initial size of messages created through this factory +// */ +// int getInitialMessagePacketSize(); +// +// /** +// * Sets the initial size of messages created through this factory. +// * +// * Value must be greater than 0. +// * +// * @param size initial size of messages created through this factory. +// */ +// void setInitialMessagePacketSize(int size); +// +// /** +// * Adds an interceptor which will be executed <em>after packets are received from the server</em>. +// * +// * @param interceptor an Interceptor +// */ +// void addInterceptor(Interceptor interceptor); +// +// /** +// * Removes an interceptor. +// * +// * @param interceptor interceptor to remove +// * +// * @return <code>true</code> if the interceptor is removed from this factory, <code>false</code> else +// */ +// boolean removeInterceptor(Interceptor interceptor); +// +// /** +// * Closes this factory and release all its resources +// */ +// void close(); - /** - * Returns whether messages will pre-acknowledged on the server before they are sent to the consumers or not. - * - * Default value is {@link HornetQClient#DEFAULT_PRE_ACKNOWLEDGE} - */ - boolean isPreAcknowledge(); - - /** - * Sets to <code>true</code> to pre-acknowledge consumed messages on the server before they are sent to consumers, else set to <code>false</code> to let - * clients acknowledge the message they consume. - * - * @param preAcknowledge <code>true</code> to enable pre-acknowledgement, <code>false</code> else - */ - void setPreAcknowledge(boolean preAcknowledge); - - /** - * Returns the acknowledgements batch size. - * - * Default value is {@link HornetQClient#DEFAULT_ACK_BATCH_SIZE}. - * - * @return the acknowledgements batch size - */ - int getAckBatchSize(); - - /** - * Sets the acknowledgements batch size. - * - * Value must be equal or greater than 0. - * - * @param ackBatchSize acknowledgements batch size - */ - void setAckBatchSize(int ackBatchSize); - - /** - * Returns the local bind address to which the multicast socket is bound for discovery. - * - * This is null if the multicast socket is not bound, or no discovery is being used - * - * @return the local bind address to which the multicast socket is bound for discovery - */ - String getLocalBindAddress(); - - /** - * Sets the local bind address to which the multicast socket is bound for discovery. - * - * @param the local bind address - */ - void setLocalBindAddress(String localBindAddress); - /** - * Returns the address to listen to discover which connectors this factory can use. - * The discovery address must be set to enable this factory to discover HornetQ servers. - * - * @return the address to listen to discover which connectors this factory can use - */ - String getDiscoveryAddress(); - - /** - * Sets the address to listen to discover which connectors this factory can use. - * - * @param discoveryAddress address to listen to discover which connectors this factory can use - */ - void setDiscoveryAddress(String discoveryAddress); - - /** - * Returns the port to listen to discover which connectors this factory can use. - * The discovery port must be set to enable this factory to discover HornetQ servers. - * - * @return the port to listen to discover which connectors this factory can use - */ - int getDiscoveryPort(); - - - /** - * Sets the port to listen to discover which connectors this factory can use. - * - * @param discoveryPort port to listen to discover which connectors this factory can use - */ - void setDiscoveryPort(int discoveryPort); - - /** - * Returns the refresh timeout for discovered HornetQ servers. - * - * If this factory uses discovery to find HornetQ servers, the list of discovered servers - * will be refreshed according to this timeout. - * - * Value is in milliseconds, default value is {@link HornetQClient#DEFAULT_DISCOVERY_REFRESH_TIMEOUT}. - * - * @return the refresh timeout for discovered HornetQ servers - */ - long getDiscoveryRefreshTimeout(); - - /** - * Sets the refresh timeout for discovered HornetQ servers. - * - * Value must be greater than 0. - * - * @param discoveryRefreshTimeout refresh timeout (in milliseconds) for discovered HornetQ servers - */ - void setDiscoveryRefreshTimeout(long discoveryRefreshTimeout); - - /** - * Returns the initial wait timeout if this factory is configured to use discovery. - * - * Value is in milliseconds, default value is {@link HornetQClient#DEFAULT_DISCOVERY_INITIAL_WAIT_TIMEOUT}. - * - * @return the initial wait timeout if this factory is configured to use discovery - */ - long getDiscoveryInitialWaitTimeout(); - - /** - * Sets the initial wait timeout if this factory is configured to use discovery. - * - * Value is in milliseconds and must be greater than 0. - * - * @param initialWaitTimeout initial wait timeout when using discovery - */ - void setDiscoveryInitialWaitTimeout(long initialWaitTimeout); - - /** - * Returns whether this factory will use global thread pools (shared among all the factories in the same JVM) - * or its own pools. - * - * Default value is {@link HornetQClient#DEFAULT_USE_GLOBAL_POOLS}. - * - * @return <code>true</code> if this factory uses global thread pools, <code>false</code> else - */ - boolean isUseGlobalPools(); - - /** - * Sets whether this factory will use global thread pools (shared among all the factories in the same JVM) - * or its own pools. - * - * @param useGlobalPools <code>true</code> to let this factory uses global thread pools, <code>false</code> else - */ - void setUseGlobalPools(boolean useGlobalPools); - - /** - * Returns the maximum size of the scheduled thread pool. - * - * Default value is {@link HornetQClient#DEFAULT_SCHEDULED_THREAD_POOL_MAX_SIZE}. - * - * @return the maximum size of the scheduled thread pool. - */ - int getScheduledThreadPoolMaxSize(); - - /** - * Sets the maximum size of the scheduled thread pool. - * - * This setting is relevant only if this factory does not use global pools. - * Value must be greater than 0. - * - * @param scheduledThreadPoolMaxSize maximum size of the scheduled thread pool. - */ - void setScheduledThreadPoolMaxSize(int scheduledThreadPoolMaxSize); - - /** - * Returns the maximum size of the thread pool. - * - * Default value is {@link HornetQClient#DEFAULT_THREAD_POOL_MAX_SIZE}. - * - * @return the maximum size of the thread pool. - */ - int getThreadPoolMaxSize(); - - /** - * Sets the maximum size of the thread pool. - * - * This setting is relevant only if this factory does not use global pools. - * Value must be -1 (for unlimited thread pool) or greater than 0. - * - * @param threadPoolMaxSize maximum size of the thread pool. - */ - void setThreadPoolMaxSize(int threadPoolMaxSize); - - /** - * Returns the time to retry connections created by this factory after failure. - * - * Value is in milliseconds, default is {@link HornetQClient#DEFAULT_RETRY_INTERVAL}. - * - * @return the time to retry connections created by this factory after failure - */ - long getRetryInterval(); - - /** - * Sets the time to retry connections created by this factory after failure. - * - * Value must be greater than 0. - * - * @param retryInterval time (in milliseconds) to retry connections created by this factory after failure - */ - void setRetryInterval(long retryInterval); - - /** - * Returns the multiplier to apply to successive retry intervals. - * - * Default value is {@link HornetQClient#DEFAULT_RETRY_INTERVAL_MULTIPLIER}. - * - * @return the multiplier to apply to successive retry intervals - */ - double getRetryIntervalMultiplier(); - - /** - * Sets the multiplier to apply to successive retry intervals. - * - * Value must be positive. - * - * @param retryIntervalMultiplier multiplier to apply to successive retry intervals - */ - void setRetryIntervalMultiplier(double retryIntervalMultiplier); - - /** - * Returns the maximum retry interval (in the case a retry interval multiplier has been specified). - * - * Value is in milliseconds, default value is {@link HornetQClient#DEFAULT_MAX_RETRY_INTERVAL}. - * - * @return the maximum retry interval - */ - long getMaxRetryInterval(); - - /** - * Sets the maximum retry interval. - * - * Value must be greater than 0. - * - * @param maxRetryInterval maximum retry interval to apply in the case a retry interval multiplier has been specified - */ - void setMaxRetryInterval(long maxRetryInterval); - - /** - * Returns the maximum number of attempts to retry connection in case of failure. - * - * Default value is {@link HornetQClient#DEFAULT_RECONNECT_ATTEMPTS}. - * - * @return the maximum number of attempts to retry connection in case of failure. - */ - int getReconnectAttempts(); - - /** - * Sets the maximum number of attempts to retry connection in case of failure. - * - * Value must be -1 (to retry infinitely), 0 (to never retry connection) or greater than 0. - * - * @param reconnectAttempts maximum number of attempts to retry connection in case of failure - */ - void setReconnectAttempts(int reconnectAttempts); +// List<Pair<TransportConfiguration, TransportConfiguration>> getClusterTopology(); +// +// void registerTopologyListener(ClusterTopologyListener listener); +// +// void unregisterTopologyListener(ClusterTopologyListener listener); - /** - * Returns true if the client will automatically attempt to connect to the backup server if the initial - * connection to the live server fails - * - * Default value is {@link HornetQClient.DEFAULT_FAILOVER_ON_INITIAL_CONNECTION}. - * - * @return - */ - boolean isFailoverOnInitialConnection(); - - /** - * Sets the value for FailoverOnInitialReconnection - * - * @param failover - */ - void setFailoverOnInitialConnection(boolean failover); - - /** - * Returns whether connections created by this factory must failover in case the server they are - * connected to <em>has normally shut down</em>. - * - * Default value is {@link HornetQClient#DEFAULT_FAILOVER_ON_SERVER_SHUTDOWN}. - * - * @return <code>true</code> if connections must failover if the server has normally shut down, else <code>false</code> - */ - boolean isFailoverOnServerShutdown(); - - /** - * Sets whether connections created by this factory must failover in case the server they are - * connected to <em>has normally shut down</em> - * - * @param failoverOnServerShutdown <code>true</code> if connections must failover if the server has normally shut down, <code>false</code> else - */ - void setFailoverOnServerShutdown(boolean failoverOnServerShutdown); - - /** - * Returns the class name of the connection load balancing policy. - * - * Default value is "org.hornetq.api.core.client.loadbalance.RoundRobinConnectionLoadBalancingPolicy". - * - * @return the class name of the connection load balancing policy - */ - String getConnectionLoadBalancingPolicyClassName(); - - /** - * Sets the class name of the connection load balancing policy. - * - * Value must be the name of a class implementing {@link ConnectionLoadBalancingPolicy}. - * - * @param loadBalancingPolicyClassName class name of the connection load balancing policy - */ - void setConnectionLoadBalancingPolicyClassName(String loadBalancingPolicyClassName); - - /** - * Returns the initial size of messages created through this factory. - * - * Value is in bytes, default value is {@link HornetQClient#DEFAULT_INITIAL_MESSAGE_PACKET_SIZE}. - * - * @return the initial size of messages created through this factory - */ - int getInitialMessagePacketSize(); - - /** - * Sets the initial size of messages created through this factory. - * - * Value must be greater than 0. - * - * @param size initial size of messages created through this factory. - */ - void setInitialMessagePacketSize(int size); - - /** - * Adds an interceptor which will be executed <em>after packets are received from the server</em>. - * - * @param interceptor an Interceptor - */ - void addInterceptor(Interceptor interceptor); - - /** - * Removes an interceptor. - * - * @param interceptor interceptor to remove - * - * @return <code>true</code> if the interceptor is removed from this factory, <code>false</code> else - */ - boolean removeInterceptor(Interceptor interceptor); - - /** - * Closes this factory and release all its resources - */ void close(); - /** - * Creates a copy of this factory. - * - * @return a copy of this factory with the same parameters values - */ - ClientSessionFactory copy(); - } Modified: branches/2_2_0_HA_Improvements/src/main/org/hornetq/api/core/client/HornetQClient.java =================================================================== --- branches/2_2_0_HA_Improvements/src/main/org/hornetq/api/core/client/HornetQClient.java 2010-06-11 04:11:42 UTC (rev 9306) +++ branches/2_2_0_HA_Improvements/src/main/org/hornetq/api/core/client/HornetQClient.java 2010-06-11 15:33:36 UTC (rev 9307) @@ -16,6 +16,7 @@ import org.hornetq.api.core.TransportConfiguration; import org.hornetq.api.core.client.loadbalance.RoundRobinConnectionLoadBalancingPolicy; import org.hornetq.core.client.impl.ClientSessionFactoryImpl; +import org.hornetq.core.client.impl.ServerLocatorImpl; import java.util.List; @@ -93,73 +94,61 @@ public static final int DEFAULT_INITIAL_MESSAGE_PACKET_SIZE = 1500; /** - * Creates a ClientSessionFactory using all the defaults. - * - * @return the ClientSessionFactory. + * Create a ServerLocator which creates session factories using a static list of transportConfigurations, the ServerLocator is not updated automatically + * as the cluster topology changes, and no HA backup information is propagated to the client + * + * @param transportConfigurations + * @return the ServerLocator */ - public static ClientSessionFactory createClientSessionFactory() + public static ServerLocator createServerLocatorWithoutHA(TransportConfiguration... transportConfigurations) { - return new ClientSessionFactoryImpl(); + return new ServerLocatorImpl(false, transportConfigurations); } - + /** - * Creates a new ClientSessionFactory using the same configuration as the one passed in. - * - * @param other The ClientSessionFactory to copy - * @return The new ClientSessionFactory + * Create a ServerLocator which creates session factories from a set of live servers, no HA backup information is propagated to the client + * + * The UDP address and port are used to listen for live servers in the cluster + * + * @param discoveryAddress The UDP group address to listen for updates + * @param discoveryPort the UDP port to listen for updates + * @return the ServerLocator */ - public static ClientSessionFactory createClientSessionFactory(final ClientSessionFactory other) + public static ServerLocator createServerLocatorWithoutHA(String discoveryAddress, final int discoveryPort) { - return new ClientSessionFactoryImpl(other); + return new ServerLocatorImpl(false, discoveryAddress, discoveryPort); } - + /** - * Creates a ClientSessionFactory that uses discovery to connect to the servers. - * - * @param discoveryAddress The address to use for discovery - * @param discoveryPort The port to use for discovery. - * @return The ClientSessionFactory. + * Create a ServerLocator which will receive cluster topology updates from the cluster as servers leave or join and new backups are appointed or removed. + * The initial list of servers supplied in this method is simply to make an initial connection to the cluster, once that connection is made, up to date + * cluster topology information is downloaded and automatically updated whenever the cluster topology changes. If the topology includes backup servers + * that information is also propagated to the client so that it can know which server to failover onto in case of live server failure. + * @param initialServers The initial set of servers used to make a connection to the cluster. Each one is tried in turn until a successful connection is made. Once + * a connection is made, the cluster topology is downloaded and the rest of the list is ignored. + * @return the ServerLocator */ - public static ClientSessionFactory createClientSessionFactory(final String discoveryAddress, final int discoveryPort) + public static ServerLocator createServerLocatorWithHA(TransportConfiguration... initialServers) { - return new ClientSessionFactoryImpl(discoveryAddress, discoveryPort); + return new ServerLocatorImpl(true, initialServers); } - + /** - * Creates a ClientSessionFactory using a List of TransportConfigurations and backups. - * - * @param staticConnectors The list of TransportConfiguration's to use. - * @return The ClientSessionFactory. + * Create a ServerLocator which will receive cluster topology updates from the cluster as servers leave or join and new backups are appointed or removed. + * The discoveryAddress and discoveryPort parameters in this method are used to listen for UDP broadcasts which contain connection information for members of the cluster. + * The broadcasted connection information is simply used to make an initial connection to the cluster, once that connection is made, up to date + * cluster topology information is downloaded and automatically updated whenever the cluster topology changes. If the topology includes backup servers + * that information is also propagated to the client so that it can know which server to failover onto in case of live server failure. + * @param discoveryAddress The UDP group address to listen for updates + * @param discoveryPort the UDP port to listen for updates + * @return the ServerLocator */ - public static ClientSessionFactory createClientSessionFactory(final List<Pair<TransportConfiguration, TransportConfiguration>> staticConnectors) + public static ServerLocator createServerLocatorWithHA(String discoveryAddress, final int discoveryPort) { - return new ClientSessionFactoryImpl(staticConnectors); + return new ServerLocatorImpl(true, discoveryAddress, discoveryPort); } + - /** - * Creates a ClientConnectionFactory using a TransportConfiguration of the server and a backup if needed. - * - * @param connectorConfig The TransportConfiguration of the server to connect to. - * @param backupConnectorConfig The TransportConfiguration of the backup server to connect to (or {@code null} if there is no backup) - * @return The ClientSessionFactory. - */ - public static ClientSessionFactory createClientSessionFactory(final TransportConfiguration connectorConfig, - final TransportConfiguration backupConnectorConfig) - { - return new ClientSessionFactoryImpl(connectorConfig, backupConnectorConfig); - } - - /** - * Creates a ClientSessionFactory using the TransportConfiguration of the server to connect to. - * - * @param connectorConfig The TransportConfiguration of the server. - * @return The ClientSessionFactory. - */ - public static ClientSessionFactory createClientSessionFactory(final TransportConfiguration connectorConfig) - { - return new ClientSessionFactoryImpl(connectorConfig); - } - private HornetQClient() { } Modified: branches/2_2_0_HA_Improvements/src/main/org/hornetq/api/core/management/HornetQServerControl.java =================================================================== --- branches/2_2_0_HA_Improvements/src/main/org/hornetq/api/core/management/HornetQServerControl.java 2010-06-11 04:11:42 UTC (rev 9306) +++ branches/2_2_0_HA_Improvements/src/main/org/hornetq/api/core/management/HornetQServerControl.java 2010-06-11 15:33:36 UTC (rev 9307) @@ -29,11 +29,11 @@ // Attributes ---------------------------------------------------- /** - * Returns the name of the connector used to connect to the backup. + * Returns the name of the connector used to connect to the live. * <br> - * If this server has no backup or is itself a backup, the value is {@code null}. + * If this server is not a backup that uses shared nothing HA, it's value is null */ - String getBackupConnectorName(); + String getLiveConnectorName(); /** * Returns this server's version. Modified: branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java =================================================================== --- branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java 2010-06-11 04:11:42 UTC (rev 9306) +++ branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java 2010-06-11 15:33:36 UTC (rev 9307) @@ -1,5 +1,5 @@ /* - * Copyright 2009 Red Hat, Inc. + * Copyright 2010 Red Hat, Inc. * Red Hat 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 @@ -10,52 +10,66 @@ * implied. See the License for the specific language governing * permissions and limitations under the License. */ + package org.hornetq.core.client.impl; -import java.io.Serializable; -import java.net.InetAddress; -import java.security.AccessController; -import java.security.PrivilegedAction; +import java.lang.ref.WeakReference; import java.util.ArrayList; import java.util.HashSet; -import java.util.Iterator; -import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.Set; -import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.Executor; import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; +import java.util.concurrent.Future; import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.ThreadFactory; import java.util.concurrent.TimeUnit; +import java.util.concurrent.locks.Lock; +import org.hornetq.api.core.HornetQBuffer; import org.hornetq.api.core.HornetQException; import org.hornetq.api.core.Interceptor; -import org.hornetq.api.core.Pair; import org.hornetq.api.core.TransportConfiguration; import org.hornetq.api.core.client.ClientSession; -import org.hornetq.api.core.client.ClientSessionFactory; -import org.hornetq.api.core.client.HornetQClient; -import org.hornetq.api.core.client.loadbalance.ConnectionLoadBalancingPolicy; -import org.hornetq.core.cluster.DiscoveryEntry; -import org.hornetq.core.cluster.DiscoveryGroup; -import org.hornetq.core.cluster.DiscoveryListener; -import org.hornetq.core.cluster.impl.DiscoveryGroupImpl; +import org.hornetq.api.core.client.SessionFailureListener; import org.hornetq.core.logging.Logger; -import org.hornetq.utils.HornetQThreadFactory; +import org.hornetq.core.protocol.core.Channel; +import org.hornetq.core.protocol.core.ChannelHandler; +import org.hornetq.core.protocol.core.CoreRemotingConnection; +import org.hornetq.core.protocol.core.Packet; +import org.hornetq.core.protocol.core.impl.PacketImpl; +import org.hornetq.core.protocol.core.impl.RemotingConnectionImpl; +import org.hornetq.core.protocol.core.impl.wireformat.ClusterTopologyMessage; +import org.hornetq.core.protocol.core.impl.wireformat.CreateSessionMessage; +import org.hornetq.core.protocol.core.impl.wireformat.CreateSessionResponseMessage; +import org.hornetq.core.protocol.core.impl.wireformat.Ping; +import org.hornetq.core.remoting.FailureListener; +import org.hornetq.core.version.Version; +import org.hornetq.spi.core.protocol.ProtocolType; +import org.hornetq.spi.core.remoting.BufferHandler; +import org.hornetq.spi.core.remoting.Connection; +import org.hornetq.spi.core.remoting.ConnectionLifeCycleListener; +import org.hornetq.spi.core.remoting.Connector; +import org.hornetq.spi.core.remoting.ConnectorFactory; +import org.hornetq.utils.ConcurrentHashSet; +import org.hornetq.utils.ConfigurationHelper; +import org.hornetq.utils.ExecutorFactory; +import org.hornetq.utils.OrderedExecutorFactory; import org.hornetq.utils.UUIDGenerator; +import org.hornetq.utils.VersionLoader; /** - * @author <a href="mailto:tim.fox@jboss.com">Tim Fox</a> - * @author <a href="mailto:clebert.suconic@jboss.org">Clebert Suconic</a> - * @author <a href="mailto:jmesnil@redhat.com">Jeff Mesnil</a> - * @author <a href="mailto:ataylor@redhat.com">Andy Taylor</a> - * @version <tt>$Revision: 3602 $</tt> + * A ClientSessionFactoryImpl * + * Encapsulates a connection to a server + * + * @author Tim Fox + * + * */ -public class ClientSessionFactoryImpl implements ClientSessionFactoryInternal, DiscoveryListener, Serializable +public class ClientSessionFactoryImpl implements ClientSessionFactoryInternal, ConnectionLifeCycleListener { + // Constants // ------------------------------------------------------------------------------------ @@ -66,1130 +80,1160 @@ // Attributes // ----------------------------------------------------------------------------------- - private final Map<Pair<TransportConfiguration, TransportConfiguration>, FailoverManager> failoverManagerMap = new LinkedHashMap<Pair<TransportConfiguration, TransportConfiguration>, FailoverManager>(); + private final ServerLocatorInternal serverLocator; - private volatile boolean receivedBroadcast = false; + private TransportConfiguration connectorConfig; - private ExecutorService threadPool; + private ConnectorFactory connectorFactory; - private ScheduledExecutorService scheduledThreadPool; + private Map<String, Object> transportParams; - private DiscoveryGroup discoveryGroup; + private final long callTimeout; - private ConnectionLoadBalancingPolicy loadBalancingPolicy; + private final long clientFailureCheckPeriod; - private FailoverManager[] failoverManagerArray; + private final long connectionTTL; - private boolean readOnly; + private final Set<ClientSessionInternal> sessions = new HashSet<ClientSessionInternal>(); - // Settable attributes: + private final Object exitLock = new Object(); - private boolean cacheLargeMessagesClient = HornetQClient.DEFAULT_CACHE_LARGE_MESSAGE_CLIENT; + private final Object createSessionLock = new Object(); - private List<Pair<TransportConfiguration, TransportConfiguration>> staticConnectors; + private boolean inCreateSession; - private String localBindAddress; + private final Object failoverLock = new Object(); - private String discoveryAddress; + private final ExecutorFactory orderedExecutorFactory; - private int discoveryPort; + private final ExecutorService threadPool; - private long discoveryRefreshTimeout; + private final ScheduledExecutorService scheduledThreadPool; - private long discoveryInitialWaitTimeout; + private final Executor closeExecutor; - private long clientFailureCheckPeriod; + private CoreRemotingConnection connection; - private long connectionTTL; + private final long retryInterval; - private long callTimeout; + private final double retryIntervalMultiplier; // For exponential backoff - private int minLargeMessageSize; + private final long maxRetryInterval; - private int consumerWindowSize; + private final int reconnectAttempts; - private int consumerMaxRate; + private final boolean failoverOnServerShutdown; - private int confirmationWindowSize; + private final Set<SessionFailureListener> listeners = new ConcurrentHashSet<SessionFailureListener>(); - private int producerWindowSize; + private Connector connector; - private int producerMaxRate; + private Future<?> pingerFuture; - private boolean blockOnAcknowledge; + private PingRunnable pingRunnable; - private boolean blockOnDurableSend; + private volatile boolean exitLoop; - private boolean blockOnNonDurableSend; + private final List<Interceptor> interceptors; - private boolean autoGroup; + private volatile boolean stopPingingAfterOne; - private boolean preAcknowledge; + private volatile boolean closed; - private String connectionLoadBalancingPolicyClassName; + // Static + // --------------------------------------------------------------------------------------- - private int ackBatchSize; + // Constructors + // --------------------------------------------------------------------------------- - private boolean useGlobalPools; + public ClientSessionFactoryImpl(final ServerLocatorInternal serverLocator, + final TransportConfiguration connectorConfig, + final boolean failoverOnServerShutdown, + final long callTimeout, + final long clientFailureCheckPeriod, + final long connectionTTL, + final long retryInterval, + final double retryIntervalMultiplier, + final long maxRetryInterval, + final int reconnectAttempts, + final boolean failoverOnInitialConnection, + final ExecutorService threadPool, + final ScheduledExecutorService scheduledThreadPool, + final List<Interceptor> interceptors) throws HornetQException + { + this.serverLocator = serverLocator; - private int scheduledThreadPoolMaxSize; + this.connectorConfig = connectorConfig; - private int threadPoolMaxSize; + this.failoverOnServerShutdown = failoverOnServerShutdown; - private long retryInterval; + connectorFactory = instantiateConnectorFactory(connectorConfig.getFactoryClassName()); - private double retryIntervalMultiplier; + transportParams = connectorConfig.getParams(); - private long maxRetryInterval; + checkTransportKeys(connectorFactory, transportParams); - private int reconnectAttempts; + this.callTimeout = callTimeout; - private boolean failoverOnInitialConnection; + this.clientFailureCheckPeriod = clientFailureCheckPeriod; - private int initialMessagePacketSize; + this.connectionTTL = connectionTTL; - private volatile boolean closed; + this.retryInterval = retryInterval; - private boolean failoverOnServerShutdown; + this.retryIntervalMultiplier = retryIntervalMultiplier; - private final List<Interceptor> interceptors = new CopyOnWriteArrayList<Interceptor>(); + this.maxRetryInterval = maxRetryInterval; - private static ExecutorService globalThreadPool; + this.reconnectAttempts = reconnectAttempts; - private static ScheduledExecutorService globalScheduledThreadPool; + this.scheduledThreadPool = scheduledThreadPool; - private String groupID; + this.threadPool = threadPool; - private static synchronized ExecutorService getGlobalThreadPool() - { - if (ClientSessionFactoryImpl.globalThreadPool == null) - { - ThreadFactory factory = new HornetQThreadFactory("HornetQ-client-global-threads", true, getThisClassLoader()); + orderedExecutorFactory = new OrderedExecutorFactory(threadPool); - ClientSessionFactoryImpl.globalThreadPool = Executors.newCachedThreadPool(factory); - } + closeExecutor = orderedExecutorFactory.getExecutor(); - return ClientSessionFactoryImpl.globalThreadPool; - } + this.interceptors = interceptors; - private static synchronized ScheduledExecutorService getGlobalScheduledThreadPool() - { - if (ClientSessionFactoryImpl.globalScheduledThreadPool == null) + // Get the connection + + getConnectionWithRetry(reconnectAttempts); + + if (connection == null && failoverOnInitialConnection) { - ThreadFactory factory = new HornetQThreadFactory("HornetQ-client-global-scheduled-threads", true, getThisClassLoader()); + TransportConfiguration backupConfig = serverLocator.getBackup(connectorConfig); - ClientSessionFactoryImpl.globalScheduledThreadPool = Executors.newScheduledThreadPool(HornetQClient.DEFAULT_SCHEDULED_THREAD_POOL_MAX_SIZE, + if (backupConfig != null) + { + // Try and connect to the backup - factory); + log.warn("Server is not available to make initial connection to. Will try backup server instead."); + + this.connectorConfig = backupConfig; + + connectorFactory = instantiateConnectorFactory(connectorConfig.getFactoryClassName()); + + transportParams = connectorConfig.getParams(); + + getConnectionWithRetry(reconnectAttempts); + } } - return ClientSessionFactoryImpl.globalScheduledThreadPool; + if (connection == null) + { + throw new HornetQException(HornetQException.NOT_CONNECTED, + "Unable to connect to server using configuration " + connectorConfig); + } + } - private void setThreadPools() + public ClientSession createSession(final String username, + final String password, + final boolean xa, + final boolean autoCommitSends, + final boolean autoCommitAcks, + final boolean preAcknowledge, + final int ackBatchSize) throws HornetQException { - if (useGlobalPools) - { - threadPool = ClientSessionFactoryImpl.getGlobalThreadPool(); + return createSessionInternal(username, + password, + xa, + autoCommitSends, + autoCommitAcks, + preAcknowledge, + ackBatchSize); + } - scheduledThreadPool = ClientSessionFactoryImpl.getGlobalScheduledThreadPool(); - } - else - { - ThreadFactory factory = new HornetQThreadFactory("HornetQ-client-factory-threads-" + System.identityHashCode(this), - true, getThisClassLoader()); + public ClientSession createSession(final boolean autoCommitSends, + final boolean autoCommitAcks, + final int ackBatchSize) throws HornetQException + { + return createSessionInternal(null, + null, + false, + autoCommitSends, + autoCommitAcks, + serverLocator.isPreAcknowledge(), + ackBatchSize); + } - if (threadPoolMaxSize == -1) - { - threadPool = Executors.newCachedThreadPool(factory); - } - else - { - threadPool = Executors.newFixedThreadPool(threadPoolMaxSize, factory); - } + public ClientSession createXASession() throws HornetQException + { + return createSessionInternal(null, + null, + true, + false, + false, + serverLocator.isPreAcknowledge(), + serverLocator.getAckBatchSize()); + } - factory = new HornetQThreadFactory("HornetQ-client-factory-pinger-threads-" + System.identityHashCode(this), - true, getThisClassLoader()); + public ClientSession createTransactedSession() throws HornetQException + { + return createSessionInternal(null, + null, + false, + false, + false, + serverLocator.isPreAcknowledge(), + serverLocator.getAckBatchSize()); + } - scheduledThreadPool = Executors.newScheduledThreadPool(scheduledThreadPoolMaxSize, factory); - } + public ClientSession createSession() throws HornetQException + { + return createSessionInternal(null, + null, + false, + true, + true, + serverLocator.isPreAcknowledge(), + serverLocator.getAckBatchSize()); } - private synchronized void initialise() throws Exception + public ClientSession createSession(final boolean autoCommitSends, final boolean autoCommitAcks) throws HornetQException { - if (!readOnly) - { - setThreadPools(); + return createSessionInternal(null, + null, + false, + autoCommitSends, + autoCommitAcks, + serverLocator.isPreAcknowledge(), + serverLocator.getAckBatchSize()); + } - instantiateLoadBalancingPolicy(); + public ClientSession createSession(final boolean xa, final boolean autoCommitSends, final boolean autoCommitAcks) throws HornetQException + { + return createSessionInternal(null, + null, + xa, + autoCommitSends, + autoCommitAcks, + serverLocator.isPreAcknowledge(), + serverLocator.getAckBatchSize()); + } - if (discoveryAddress != null) - { - InetAddress groupAddress = InetAddress.getByName(discoveryAddress); + public ClientSession createSession(final boolean xa, + final boolean autoCommitSends, + final boolean autoCommitAcks, + final boolean preAcknowledge) throws HornetQException + { + return createSessionInternal(null, + null, + xa, + autoCommitSends, + autoCommitAcks, + preAcknowledge, + serverLocator.getAckBatchSize()); + } - InetAddress lbAddress; + // ConnectionLifeCycleListener implementation -------------------------------------------------- - if (localBindAddress != null) - { - lbAddress = InetAddress.getByName(localBindAddress); - } - else - { - lbAddress = null; - } + public void connectionCreated(final Connection connection, final ProtocolType protocol) + { + } - discoveryGroup = new DiscoveryGroupImpl(UUIDGenerator.getInstance().generateStringUUID(), - discoveryAddress, - lbAddress, - groupAddress, - discoveryPort, - discoveryRefreshTimeout); + public void connectionDestroyed(final Object connectionID) + { + handleConnectionFailure(connectionID, + new HornetQException(HornetQException.NOT_CONNECTED, "Channel disconnected")); + } - discoveryGroup.registerListener(this); + public void connectionException(final Object connectionID, final HornetQException me) + { + handleConnectionFailure(connectionID, me); + } - discoveryGroup.start(); - } - else if (staticConnectors != null) + // Must be synchronized to prevent it happening concurrently with failover which can lead to + // inconsistencies + public void removeSession(final ClientSessionInternal session) + { + synchronized (createSessionLock) + { + synchronized (failoverLock) { - for (Pair<TransportConfiguration, TransportConfiguration> pair : staticConnectors) - { - FailoverManager cm = new FailoverManagerImpl(this, - pair.a, - pair.b, - failoverOnServerShutdown, - callTimeout, - clientFailureCheckPeriod, - connectionTTL, - retryInterval, - retryIntervalMultiplier, - maxRetryInterval, - reconnectAttempts, - failoverOnInitialConnection, - threadPool, - scheduledThreadPool, - interceptors); + sessions.remove(session); - failoverManagerMap.put(pair, cm); - } - - updatefailoverManagerArray(); + checkCloseConnection(); } - else - { - throw new IllegalStateException("Before using a session factory you must either set discovery address and port or " + "provide some static transport configuration"); - } - readOnly = true; } } - // Static - // --------------------------------------------------------------------------------------- + public synchronized int numConnections() + { + return connection != null ? 1 : 0; + } - // Constructors - // --------------------------------------------------------------------------------- + public int numSessions() + { + return sessions.size(); + } - public ClientSessionFactoryImpl(final ClientSessionFactory other) + public void addFailureListener(final SessionFailureListener listener) { - localBindAddress = other.getLocalBindAddress(); + listeners.add(listener); + } - discoveryAddress = other.getDiscoveryAddress(); + public boolean removeFailureListener(final SessionFailureListener listener) + { + return listeners.remove(listener); + } - discoveryPort = other.getDiscoveryPort(); + public void causeExit() + { + exitLoop = true; + } - staticConnectors = other.getStaticConnectors(); + public void close() + { + if (closed) + { + return; + } - discoveryRefreshTimeout = other.getDiscoveryRefreshTimeout(); + causeExit(); - clientFailureCheckPeriod = other.getClientFailureCheckPeriod(); + closed = true; + } - connectionTTL = other.getConnectionTTL(); + // Public + // --------------------------------------------------------------------------------------- - callTimeout = other.getCallTimeout(); + public void stopPingingAfterOne() + { + stopPingingAfterOne = true; + } - minLargeMessageSize = other.getMinLargeMessageSize(); + // Protected + // ------------------------------------------------------------------------------------ - consumerWindowSize = other.getConsumerWindowSize(); + // Package Private + // ------------------------------------------------------------------------------ - consumerMaxRate = other.getConsumerMaxRate(); + // Private + // -------------------------------------------------------------------------------------- - confirmationWindowSize = other.getConfirmationWindowSize(); + private void handleConnectionFailure(final Object connectionID, final HornetQException me) + { + failoverOrReconnect(connectionID, me); + } - producerWindowSize = other.getProducerWindowSize(); + private void failoverOrReconnect(final Object connectionID, final HornetQException me) + { + Set<ClientSessionInternal> sessionsToClose = null; - producerMaxRate = other.getProducerMaxRate(); + synchronized (failoverLock) + { + if (connection == null || connection.getID() != connectionID) + { + // We already failed over/reconnected - probably the first failure came in, all the connections were failed + // over then a async connection exception or disconnect + // came in for one of the already exitLoop connections, so we return true - we don't want to call the + // listeners again - blockOnAcknowledge = other.isBlockOnAcknowledge(); + return; + } - blockOnDurableSend = other.isBlockOnDurableSend(); + // We call before reconnection occurs to give the user a chance to do cleanup, like cancel messages + callFailureListeners(me, false); - blockOnNonDurableSend = other.isBlockOnNonDurableSend(); + // Now get locks on all channel 1s, whilst holding the failoverLock - this makes sure + // There are either no threads executing in createSession, or one is blocking on a createSession + // result. - autoGroup = other.isAutoGroup(); + // Then interrupt the channel 1 that is blocking (could just interrupt them all) - preAcknowledge = other.isPreAcknowledge(); + // Then release all channel 1 locks - this allows the createSession to exit the monitor - ackBatchSize = other.getAckBatchSize(); + // Then get all channel 1 locks again - this ensures the any createSession thread has executed the section and + // returned all its connections to the connection manager (the code to return connections to connection manager + // must be inside the lock - connectionLoadBalancingPolicyClassName = other.getConnectionLoadBalancingPolicyClassName(); + // Then perform failover - discoveryInitialWaitTimeout = other.getDiscoveryInitialWaitTimeout(); + // Then release failoverLock - useGlobalPools = other.isUseGlobalPools(); + // The other side of the bargain - during createSession: + // The calling thread must get the failoverLock and get its' connections when this is locked. + // While this is still locked it must then get the channel1 lock + // It can then release the failoverLock + // It should catch HornetQException.INTERRUPTED in the call to channel.sendBlocking + // It should then return its connections, with channel 1 lock still held + // It can then release the channel 1 lock, and retry (which will cause locking on failoverLock + // until failover is complete - scheduledThreadPoolMaxSize = other.getScheduledThreadPoolMaxSize(); + boolean serverShutdown = me.getCode() == HornetQException.DISCONNECTED; - threadPoolMaxSize = other.getThreadPoolMaxSize(); + // We will try to failover if there is a backup connector factory, but we don't do this if the server + // has been shutdown cleanly unless failoverOnServerShutdown is true + TransportConfiguration backupConfig = serverLocator.getBackup(connectorConfig); + + boolean attemptFailover = backupConfig != null && (failoverOnServerShutdown || !serverShutdown); - retryInterval = other.getRetryInterval(); + boolean attemptReconnect; - retryIntervalMultiplier = other.getRetryIntervalMultiplier(); + if (attemptFailover) + { + attemptReconnect = false; + } + else + { + attemptReconnect = reconnectAttempts != 0; + } - maxRetryInterval = other.getMaxRetryInterval(); + if (attemptFailover || attemptReconnect) + { + lockChannel1(); - reconnectAttempts = other.getReconnectAttempts(); + final boolean needToInterrupt; - failoverOnInitialConnection = other.isFailoverOnInitialConnection(); + synchronized (exitLock) + { + needToInterrupt = inCreateSession; + } - failoverOnServerShutdown = other.isFailoverOnServerShutdown(); + unlockChannel1(); - cacheLargeMessagesClient = other.isCacheLargeMessagesClient(); + if (needToInterrupt) + { + // Forcing return all channels won't guarantee that any blocked thread will return immediately + // So we need to wait for it + forceReturnChannel1(); - initialMessagePacketSize = other.getInitialMessagePacketSize(); + // Now we need to make sure that the thread has actually exited and returned it's connections + // before failover occurs - groupID = other.getGroupID(); - } + synchronized (exitLock) + { + while (inCreateSession) + { + try + { + exitLock.wait(5000); + } + catch (InterruptedException e) + { + } + } + } + } - public ClientSessionFactoryImpl() - { - discoveryRefreshTimeout = HornetQClient.DEFAULT_DISCOVERY_REFRESH_TIMEOUT; + // Now we absolutely know that no threads are executing in or blocked in createSession, and no + // more will execute it until failover is complete - clientFailureCheckPeriod = HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD; + // So.. do failover / reconnection - connectionTTL = HornetQClient.DEFAULT_CONNECTION_TTL; + CoreRemotingConnection oldConnection = connection; - callTimeout = HornetQClient.DEFAULT_CALL_TIMEOUT; + connection = null; - minLargeMessageSize = HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE; + try + { + connector.close(); + } + catch (Exception ignore) + { + } - consumerWindowSize = HornetQClient.DEFAULT_CONSUMER_WINDOW_SIZE; + cancelScheduledTasks(); - consumerMaxRate = HornetQClient.DEFAULT_CONSUMER_MAX_RATE; + connector = null; - confirmationWindowSize = HornetQClient.DEFAULT_CONFIRMATION_WINDOW_SIZE; + if (attemptFailover) + { + // Now try failing over to backup - producerWindowSize = HornetQClient.DEFAULT_PRODUCER_WINDOW_SIZE; + this.connectorConfig = backupConfig; + + connectorFactory = instantiateConnectorFactory(connectorConfig.getFactoryClassName()); - producerMaxRate = HornetQClient.DEFAULT_PRODUCER_MAX_RATE; + transportParams = connectorConfig.getParams(); - blockOnAcknowledge = HornetQClient.DEFAULT_BLOCK_ON_ACKNOWLEDGE; + reconnectSessions(oldConnection, reconnectAttempts == -1 ? -1 : reconnectAttempts + 1); + } + else + { + reconnectSessions(oldConnection, reconnectAttempts); + } - blockOnDurableSend = HornetQClient.DEFAULT_BLOCK_ON_DURABLE_SEND; + oldConnection.destroy(); + } + else + { + connection.destroy(); - blockOnNonDurableSend = HornetQClient.DEFAULT_BLOCK_ON_NON_DURABLE_SEND; + connection = null; + } - autoGroup = HornetQClient.DEFAULT_AUTO_GROUP; + callFailureListeners(me, true); - preAcknowledge = HornetQClient.DEFAULT_PRE_ACKNOWLEDGE; + if (connection == null) + { + sessionsToClose = new HashSet<ClientSessionInternal>(sessions); + } + } - ackBatchSize = HornetQClient.DEFAULT_ACK_BATCH_SIZE; + // This needs to be outside the failover lock to prevent deadlock + if (sessionsToClose != null) + { + // If connection is null it means we didn't succeed in failing over or reconnecting + // so we close all the sessions, so they will throw exceptions when attempted to be used - connectionLoadBalancingPolicyClassName = HornetQClient.DEFAULT_CONNECTION_LOAD_BALANCING_POLICY_CLASS_NAME; - - discoveryInitialWaitTimeout = HornetQClient.DEFAULT_DISCOVERY_INITIAL_WAIT_TIMEOUT; - - useGlobalPools = HornetQClient.DEFAULT_USE_GLOBAL_POOLS; - - scheduledThreadPoolMaxSize = HornetQClient.DEFAULT_SCHEDULED_THREAD_POOL_MAX_SIZE; - - threadPoolMaxSize = HornetQClient.DEFAULT_THREAD_POOL_MAX_SIZE; - - retryInterval = HornetQClient.DEFAULT_RETRY_INTERVAL; - - retryIntervalMultiplier = HornetQClient.DEFAULT_RETRY_INTERVAL_MULTIPLIER; - - maxRetryInterval = HornetQClient.DEFAULT_MAX_RETRY_INTERVAL; - - reconnectAttempts = HornetQClient.DEFAULT_RECONNECT_ATTEMPTS; - - failoverOnInitialConnection = HornetQClient.DEFAULT_FAILOVER_ON_INITIAL_CONNECTION; - - failoverOnServerShutdown = HornetQClient.DEFAULT_FAILOVER_ON_SERVER_SHUTDOWN; - - cacheLargeMessagesClient = HornetQClient.DEFAULT_CACHE_LARGE_MESSAGE_CLIENT; - - initialMessagePacketSize = HornetQClient.DEFAULT_INITIAL_MESSAGE_PACKET_SIZE; + for (ClientSessionInternal session : sessionsToClose) + { + try + { + session.cleanUp(); + } + catch (Exception e) + { + log.error("Failed to cleanup session"); + } + } + } } - public ClientSessionFactoryImpl(final String discoveryAddress, final int discoveryPort) + private ClientSession createSessionInternal(final String username, + final String password, + final boolean xa, + final boolean autoCommitSends, + final boolean autoCommitAcks, + final boolean preAcknowledge, + final int ackBatchSize) throws HornetQException { - this(); + synchronized (createSessionLock) + { + String name = UUIDGenerator.getInstance().generateStringUUID(); - this.discoveryAddress = discoveryAddress; + boolean retry = false; + do + { + Version clientVersion = VersionLoader.getVersion(); - this.discoveryPort = discoveryPort; - } + Lock lock = null; - public ClientSessionFactoryImpl(final String localBindAddress, final String discoveryAddress, final int discoveryPort) - { - this(); + try + { + Channel channel1; - this.localBindAddress = localBindAddress; + synchronized (failoverLock) + { + if (connection == null) + { + throw new IllegalStateException("Connection is null"); + } - this.discoveryAddress = discoveryAddress; + channel1 = connection.getChannel(1, -1); - this.discoveryPort = discoveryPort; - } + // Lock it - this must be done while the failoverLock is held + channel1.getLock().lock(); - public ClientSessionFactoryImpl(final List<Pair<TransportConfiguration, TransportConfiguration>> staticConnectors) - { - this(); + lock = channel1.getLock(); + } // We can now release the failoverLock - this.staticConnectors = staticConnectors; - } + // We now set a flag saying createSession is executing + synchronized (exitLock) + { + inCreateSession = true; + } - public ClientSessionFactoryImpl(final TransportConfiguration connectorConfig, - final TransportConfiguration backupConnectorConfig) - { - this(); + long sessionChannelID = connection.generateChannelID(); - staticConnectors = new ArrayList<Pair<TransportConfiguration, TransportConfiguration>>(); + Packet request = new CreateSessionMessage(name, + sessionChannelID, + clientVersion.getIncrementingVersion(), + username, + password, + serverLocator.getMinLargeMessageSize(), + xa, + autoCommitSends, + autoCommitAcks, + preAcknowledge, + serverLocator.getConfirmationWindowSize(), + null); - staticConnectors.add(new Pair<TransportConfiguration, TransportConfiguration>(connectorConfig, - backupConnectorConfig)); - } + Packet pResponse; + try + { + pResponse = channel1.sendBlocking(request); + } + catch (HornetQException e) + { + if (e.getCode() == HornetQException.INCOMPATIBLE_CLIENT_SERVER_VERSIONS) + { + connection.destroy(); + } - public ClientSessionFactoryImpl(final TransportConfiguration connectorConfig) - { - this(connectorConfig, null); - } + if (e.getCode() == HornetQException.UNBLOCKED) + { + // This means the thread was blocked on create session and failover unblocked it + // so failover could occur - // ClientSessionFactory implementation------------------------------------------------------------ + retry = true; - public synchronized boolean isCacheLargeMessagesClient() - { - return cacheLargeMessagesClient; - } + continue; + } + else + { + throw e; + } + } - public synchronized void setCacheLargeMessagesClient(final boolean cached) - { - cacheLargeMessagesClient = cached; - } + CreateSessionResponseMessage response = (CreateSessionResponseMessage)pResponse; - public synchronized List<Pair<TransportConfiguration, TransportConfiguration>> getStaticConnectors() - { - return staticConnectors; - } + Channel sessionChannel = connection.getChannel(sessionChannelID, + serverLocator.getConfirmationWindowSize()); - public synchronized void setStaticConnectors(final List<Pair<TransportConfiguration, TransportConfiguration>> staticConnectors) - { - checkWrite(); + ClientSessionInternal session = new ClientSessionImpl(this, + name, + username, + password, + xa, + autoCommitSends, + autoCommitAcks, + preAcknowledge, + serverLocator.isBlockOnAcknowledge(), + serverLocator.isAutoGroup(), + ackBatchSize, + serverLocator.getConsumerWindowSize(), + serverLocator.getConsumerMaxRate(), + serverLocator.getConfirmationWindowSize(), + serverLocator.getProducerWindowSize(), + serverLocator.getProducerMaxRate(), + serverLocator.isBlockOnNonDurableSend(), + serverLocator.isBlockOnDurableSend(), + serverLocator.isCacheLargeMessagesClient(), + serverLocator.getMinLargeMessageSize(), + serverLocator.getInitialMessagePacketSize(), + serverLocator.getGroupID(), + connection, + response.getServerVersion(), + sessionChannel, + orderedExecutorFactory.getExecutor()); - this.staticConnectors = staticConnectors; - } + sessions.add(session); - public synchronized long getClientFailureCheckPeriod() - { - return clientFailureCheckPeriod; - } + ChannelHandler handler = new ClientSessionPacketHandler(session, sessionChannel); - public synchronized void setClientFailureCheckPeriod(final long clientFailureCheckPeriod) - { - checkWrite(); - this.clientFailureCheckPeriod = clientFailureCheckPeriod; - } + sessionChannel.setHandler(handler); - public synchronized long getConnectionTTL() - { - return connectionTTL; - } + return new DelegatingSession(session); + } + catch (Throwable t) + { + if (lock != null) + { + lock.unlock(); - public synchronized void setConnectionTTL(final long connectionTTL) - { - checkWrite(); - this.connectionTTL = connectionTTL; - } + lock = null; + } - public synchronized long getCallTimeout() - { - return callTimeout; - } + if (t instanceof HornetQException) + { + throw (HornetQException)t; + } + else + { + HornetQException me = new HornetQException(HornetQException.INTERNAL_ERROR, + "Failed to create session", + t); - public synchronized void setCallTimeout(final long callTimeout) - { - checkWrite(); - this.callTimeout = callTimeout; - } + throw me; + } + } + finally + { + if (lock != null) + { + lock.unlock(); + } - public synchronized int getMinLargeMessageSize() - { - return minLargeMessageSize; - } + // Execution has finished so notify any failover thread that may be waiting for us to be done + synchronized (exitLock) + { + inCreateSession = false; - public synchronized void setMinLargeMessageSize(final int minLargeMessageSize) - { - checkWrite(); - this.minLargeMessageSize = minLargeMessageSize; - } + exitLock.notify(); + } + } + } + while (retry); + } - public synchronized int getConsumerWindowSize() - { - return consumerWindowSize; + // Should never get here + throw new IllegalStateException("Oh my God it's full of stars!"); } - public synchronized void setConsumerWindowSize(final int consumerWindowSize) + private void callFailureListeners(final HornetQException me, final boolean afterReconnect) { - checkWrite(); - this.consumerWindowSize = consumerWindowSize; - } + final List<SessionFailureListener> listenersClone = new ArrayList<SessionFailureListener>(listeners); - public synchronized int getConsumerMaxRate() - { - return consumerMaxRate; + for (final SessionFailureListener listener : listenersClone) + { + try + { + if (afterReconnect) + { + listener.connectionFailed(me); + } + else + { + listener.beforeReconnect(me); + } + } + catch (final Throwable t) + { + // Failure of one listener to execute shouldn't prevent others + // from + // executing + log.error("Failed to execute failure listener", t); + } + } } - public synchronized void setConsumerMaxRate(final int consumerMaxRate) + /* + * Re-attach sessions all pre-existing sessions to the new remoting connection + */ + private void reconnectSessions(final CoreRemotingConnection oldConnection, final int reconnectAttempts) { - checkWrite(); - this.consumerMaxRate = consumerMaxRate; - } + getConnectionWithRetry(reconnectAttempts); - public synchronized int getConfirmationWindowSize() - { - return confirmationWindowSize; - } + if (connection == null) + { + log.warn("Failed to connect to server."); - public synchronized void setConfirmationWindowSize(final int confirmationWindowSize) - { - checkWrite(); - this.confirmationWindowSize = confirmationWindowSize; - } + return; + } - public synchronized int getProducerWindowSize() - { - return producerWindowSize; - } + List<FailureListener> oldListeners = oldConnection.getFailureListeners(); - public synchronized void setProducerWindowSize(final int producerWindowSize) - { - checkWrite(); - this.producerWindowSize = producerWindowSize; - } + List<FailureListener> newListeners = new ArrayList<FailureListener>(connection.getFailureListeners()); - public synchronized int getProducerMaxRate() - { - return producerMaxRate; - } + for (FailureListener listener : oldListeners) + { + // Add all apart from the first one which is the old DelegatingFailureListener - public synchronized void setProducerMaxRate(final int producerMaxRate) - { - checkWrite(); - this.producerMaxRate = producerMaxRate; - } + if (listener instanceof DelegatingFailureListener == false) + { + newListeners.add(listener); + } + } - public synchronized boolean isBlockOnAcknowledge() - { - return blockOnAcknowledge; - } + connection.setFailureListeners(newListeners); - public synchronized void setBlockOnAcknowledge(final boolean blockOnAcknowledge) - { - checkWrite(); - this.blockOnAcknowledge = blockOnAcknowledge; + for (ClientSessionInternal session : sessions) + { + session.handleFailover(connection); + } } - public synchronized boolean isBlockOnDurableSend() + private void getConnectionWithRetry(final int reconnectAttempts) { - return blockOnDurableSend; - } + long interval = retryInterval; - public synchronized void setBlockOnDurableSend(final boolean blockOnDurableSend) - { - checkWrite(); - this.blockOnDurableSend = blockOnDurableSend; - } + int count = 0; - public synchronized boolean isBlockOnNonDurableSend() - { - return blockOnNonDurableSend; - } + while (true) + { + if (exitLoop) + { + return; + } - public synchronized void setBlockOnNonDurableSend(final boolean blockOnNonDurableSend) - { - checkWrite(); - this.blockOnNonDurableSend = blockOnNonDurableSend; - } + getConnection(); - public synchronized boolean isAutoGroup() - { - return autoGroup; - } + if (connection == null) + { + // Failed to get connection - public synchronized void setAutoGroup(final boolean autoGroup) - { - checkWrite(); - this.autoGroup = autoGroup; - } + if (reconnectAttempts != 0) + { + count++; - public synchronized boolean isPreAcknowledge() - { - return preAcknowledge; - } + if (reconnectAttempts != -1 && count == reconnectAttempts) + { + log.warn("Tried " + reconnectAttempts + " times to connect. Now giving up."); - public synchronized void setPreAcknowledge(final boolean preAcknowledge) - { - checkWrite(); - this.preAcknowledge = preAcknowledge; - } + return; + } - public synchronized int getAckBatchSize() - { - return ackBatchSize; - } + try + { + Thread.sleep(interval); + } + catch (InterruptedException ignore) + { + } - public synchronized void setAckBatchSize(final int ackBatchSize) - { - checkWrite(); - this.ackBatchSize = ackBatchSize; - } + // Exponential back-off + long newInterval = (long)(interval * retryIntervalMultiplier); - public synchronized long getDiscoveryInitialWaitTimeout() - { - return discoveryInitialWaitTimeout; - } + if (newInterval > maxRetryInterval) + { + newInterval = maxRetryInterval; + } - public synchronized void setDiscoveryInitialWaitTimeout(final long initialWaitTimeout) - { - checkWrite(); - discoveryInitialWaitTimeout = initialWaitTimeout; + interval = newInterval; + } + else + { + return; + } + } + else + { + return; + } + } } - public synchronized boolean isUseGlobalPools() + private void cancelScheduledTasks() { - return useGlobalPools; - } + if (pingerFuture != null) + { + pingRunnable.cancel(); - public synchronized void setUseGlobalPools(final boolean useGlobalPools) - { - checkWrite(); - this.useGlobalPools = useGlobalPools; - } + pingerFuture.cancel(false); - public synchronized int getScheduledThreadPoolMaxSize() - { - return scheduledThreadPoolMaxSize; - } + pingRunnable = null; - public synchronized void setScheduledThreadPoolMaxSize(final int scheduledThreadPoolMaxSize) - { - checkWrite(); - this.scheduledThreadPoolMaxSize = scheduledThreadPoolMaxSize; + pingerFuture = null; + } } - public synchronized int getThreadPoolMaxSize() + private void checkCloseConnection() { - return threadPoolMaxSize; - } + if (connection != null && sessions.size() == 0) + { + cancelScheduledTasks(); - public synchronized void setThreadPoolMaxSize(final int threadPoolMaxSize) - { - checkWrite(); - this.threadPoolMaxSize = threadPoolMaxSize; - } + try + { + connection.destroy(); + } + catch (Throwable ignore) + { + } - public synchronized long getRetryInterval() - { - return retryInterval; - } + connection = null; - public synchronized void setRetryInterval(final long retryInterval) - { - checkWrite(); - this.retryInterval = retryInterval; - } + try + { + if (connector != null) + { + connector.close(); + } + } + catch (Throwable ignore) + { + } - public synchronized long getMaxRetryInterval() - { - return maxRetryInterval; + connector = null; + } } - public synchronized void setMaxRetryInterval(final long retryInterval) + public CoreRemotingConnection getConnection() { - checkWrite(); - maxRetryInterval = retryInterval; - } + if (connection == null) + { + Connection tc = null; - public synchronized double getRetryIntervalMultiplier() - { - return retryIntervalMultiplier; - } + try + { + DelegatingBufferHandler handler = new DelegatingBufferHandler(); - public synchronized void setRetryIntervalMultiplier(final double retryIntervalMultiplier) - { - checkWrite(); - this.retryIntervalMultiplier = retryIntervalMultiplier; - } + connector = connectorFactory.createConnector(transportParams, + handler, + this, + closeExecutor, + threadPool, + scheduledThreadPool); - public synchronized int getReconnectAttempts() - { - return reconnectAttempts; - } + if (connector != null) + { + connector.start(); - public synchronized void setReconnectAttempts(final int reconnectAttempts) - { - checkWrite(); - this.reconnectAttempts = reconnectAttempts; - } + tc = connector.createConnection(); - public synchronized boolean isFailoverOnInitialConnection() - { - return this.failoverOnInitialConnection; - } + if (tc == null) + { + try + { + connector.close(); + } + catch (Throwable t) + { + } - public synchronized void setFailoverOnInitialConnection(final boolean failover) - { - checkWrite(); - this.failoverOnInitialConnection = failover; - } + connector = null; + } + } + } + catch (Exception e) + { + // Sanity catch for badly behaved remoting plugins - public synchronized boolean isFailoverOnServerShutdown() - { - return failoverOnServerShutdown; - } + log.warn("connector.create or connectorFactory.createConnector should never throw an exception, implementation is badly behaved, but we'll deal with it anyway.", + e); - public synchronized void setFailoverOnServerShutdown(final boolean failoverOnServerShutdown) - { - checkWrite(); - this.failoverOnServerShutdown = failoverOnServerShutdown; - } + if (tc != null) + { + try + { + tc.close(); + } + catch (Throwable t) + { + } + } - public synchronized String getConnectionLoadBalancingPolicyClassName() - { - return connectionLoadBalancingPolicyClassName; - } + if (connector != null) + { + try + { + connector.close(); + } + catch (Throwable t) + { + } + } - public synchronized void setConnectionLoadBalancingPolicyClassName(final String loadBalancingPolicyClassName) - { - checkWrite(); - connectionLoadBalancingPolicyClassName = loadBalancingPolicyClassName; - } + tc = null; - public synchronized String getLocalBindAddress() - { - return localBindAddress; - } + connector = null; + } - public synchronized void setLocalBindAddress(final String localBindAddress) - { - checkWrite(); - this.localBindAddress = localBindAddress; - } + if (tc == null) + { + return connection; + } - public synchronized String getDiscoveryAddress() - { - return discoveryAddress; - } + connection = new RemotingConnectionImpl(tc, callTimeout, interceptors); - public synchronized void setDiscoveryAddress(final String discoveryAddress) - { - checkWrite(); - this.discoveryAddress = discoveryAddress; - } + connection.addFailureListener(new DelegatingFailureListener(connection.getID())); - public synchronized int getDiscoveryPort() - { - return discoveryPort; - } + connection.getChannel(0, -1).setHandler(new Channel0Handler(connection)); - public synchronized void setDiscoveryPort(final int discoveryPort) - { - checkWrite(); - this.discoveryPort = discoveryPort; - } + if (clientFailureCheckPeriod != -1) + { + if (pingerFuture == null) + { + pingRunnable = new PingRunnable(); - public synchronized long getDiscoveryRefreshTimeout() - { - return discoveryRefreshTimeout; - } + pingerFuture = scheduledThreadPool.scheduleWithFixedDelay(new ActualScheduledPinger(pingRunnable), + 0, + clientFailureCheckPeriod, + TimeUnit.MILLISECONDS); + } + // send a ping every time we create a new remoting connection + // to set up its TTL on the server side + else + { + pingRunnable.run(); + } + } + } - public void addInterceptor(final Interceptor interceptor) - { - interceptors.add(interceptor); + return connection; } - public boolean removeInterceptor(final Interceptor interceptor) + private ConnectorFactory instantiateConnectorFactory(final String connectorFactoryClassName) { - return interceptors.remove(interceptor); + ClassLoader loader = Thread.currentThread().getContextClassLoader(); + try + { + Class<?> clazz = loader.loadClass(connectorFactoryClassName); + return (ConnectorFactory)clazz.newInstance(); + } + catch (Exception e) + { + throw new IllegalArgumentException("Error instantiating connector factory \"" + connectorFactoryClassName + + "\"", e); + } } - public synchronized void setDiscoveryRefreshTimeout(final long discoveryRefreshTimeout) + private void lockChannel1() { - checkWrite(); - this.discoveryRefreshTimeout = discoveryRefreshTimeout; - } + Channel channel1 = connection.getChannel(1, -1); - public synchronized int getInitialMessagePacketSize() - { - return initialMessagePacketSize; + channel1.getLock().lock(); } - public synchronized void setInitialMessagePacketSize(final int size) + private void unlockChannel1() { - checkWrite(); - initialMessagePacketSize = size; - } + Channel channel1 = connection.getChannel(1, -1); - public ClientSession createSession(final String username, - final String password, - final boolean xa, - final boolean autoCommitSends, - final boolean autoCommitAcks, - final boolean preAcknowledge, - final int ackBatchSize) throws HornetQException - { - return createSessionInternal(username, - password, - xa, - autoCommitSends, - autoCommitAcks, - preAcknowledge, - ackBatchSize); + channel1.getLock().unlock(); } - public ClientSession createSession(final boolean autoCommitSends, - final boolean autoCommitAcks, - final int ackBatchSize) throws HornetQException + private void forceReturnChannel1() { - return createSessionInternal(null, null, false, autoCommitSends, autoCommitAcks, preAcknowledge, ackBatchSize); - } + Channel channel1 = connection.getChannel(1, -1); - public ClientSession createXASession() throws HornetQException - { - return createSessionInternal(null, null, true, false, false, preAcknowledge, ackBatchSize); + channel1.returnBlocking(); } - public ClientSession createTransactedSession() throws HornetQException + private void checkTransportKeys(final ConnectorFactory factory, final Map<String, Object> params) { - return createSessionInternal(null, null, false, false, false, preAcknowledge, ackBatchSize); - } - - public ClientSession createSession() throws HornetQException - { - return createSessionInternal(null, null, false, true, true, preAcknowledge, ackBatchSize); - } - - public ClientSession createSession(final boolean autoCommitSends, final boolean autoCommitAcks) throws HornetQException - { - return createSessionInternal(null, null, false, autoCommitSends, autoCommitAcks, preAcknowledge, ackBatchSize); - } - - public ClientSession createSession(final boolean xa, final boolean autoCommitSends, final boolean autoCommitAcks) throws HornetQException - { - return createSessionInternal(null, null, xa, autoCommitSends, autoCommitAcks, preAcknowledge, ackBatchSize); - } - - public ClientSession createSession(final boolean xa, - final boolean autoCommitSends, - final boolean autoCommitAcks, - final boolean preAcknowledge) throws HornetQException - { - return createSessionInternal(null, null, xa, autoCommitSends, autoCommitAcks, preAcknowledge, ackBatchSize); - } - - public int numSessions() - { - int num = 0; - - for (FailoverManager failoverManager : failoverManagerMap.values()) + if (params != null) { - num += failoverManager.numSessions(); - } + Set<String> invalid = ConfigurationHelper.checkKeys(factory.getAllowableProperties(), params.keySet()); - return num; - } + if (!invalid.isEmpty()) + { + String msg = ConfigurationHelper.stringSetToCommaListString("The following keys are invalid for configuring a connector: ", + invalid); - public int numConnections() - { - int num = 0; + throw new IllegalStateException(msg); - for (FailoverManager failoverManager : failoverManagerMap.values()) - { - num += failoverManager.numConnections(); + } } - - return num; } - public void close() + private class Channel0Handler implements ChannelHandler { - if (closed) - { - return; - } + private final CoreRemotingConnection conn; - if (discoveryGroup != null) + private Channel0Handler(final CoreRemotingConnection conn) { - try - { - discoveryGroup.stop(); - } - catch (Exception e) - { - ClientSessionFactoryImpl.log.error("Failed to stop discovery group", e); - } + this.conn = conn; } - for (FailoverManager failoverManager : failoverManagerMap.values()) + public void handlePacket(final Packet packet) { - failoverManager.causeExit(); - } + final byte type = packet.getType(); - failoverManagerMap.clear(); - - if (!useGlobalPools) - { - if (threadPool != null) + if (type == PacketImpl.DISCONNECT) { - threadPool.shutdown(); - - try + closeExecutor.execute(new Runnable() { - if (!threadPool.awaitTermination(10000, TimeUnit.MILLISECONDS)) + // Must be executed on new thread since cannot block the netty thread for a long time and fail can + // cause reconnect loop + public void run() { - ClientSessionFactoryImpl.log.warn("Timed out waiting for pool to terminate"); + conn.fail(new HornetQException(HornetQException.DISCONNECTED, + "The connection was disconnected because of server shutdown")); } - } - catch (InterruptedException ignore) - { - } + }); } - - if (scheduledThreadPool != null) + else if (type == PacketImpl.CLUSTER_TOPOLOGY) { - scheduledThreadPool.shutdown(); - - try - { - if (!scheduledThreadPool.awaitTermination(10000, TimeUnit.MILLISECONDS)) - { - ClientSessionFactoryImpl.log.warn("Timed out waiting for scheduled pool to terminate"); - } - } - catch (InterruptedException ignore) - { - } + ClusterTopologyMessage topMessage = (ClusterTopologyMessage)packet; + + serverLocator.onTopologyChanged(topMessage.getTopology()); } } - - closed = true; } - public ClientSessionFactory copy() + private class DelegatingBufferHandler implements BufferHandler { - return new ClientSessionFactoryImpl(this); - } + public void bufferReceived(final Object connectionID, final HornetQBuffer buffer) + { + CoreRemotingConnection theConn = connection; - public void setGroupID(final String groupID) - { - this.groupID = groupID; + if (theConn != null && connectionID == theConn.getID()) + { + theConn.bufferReceived(connectionID, buffer); + } + } } - public String getGroupID() + private class DelegatingFailureListener implements FailureListener { - return groupID; - } + private final Object connectionID; - // DiscoveryListener implementation -------------------------------------------------------- - - public synchronized void connectorsChanged() - { - receivedBroadcast = true; - - Map<String, DiscoveryEntry> newConnectors = discoveryGroup.getDiscoveryEntryMap(); - - Set<Pair<TransportConfiguration, TransportConfiguration>> connectorSet = new HashSet<Pair<TransportConfiguration, TransportConfiguration>>(); - - for (DiscoveryEntry entry : newConnectors.values()) + DelegatingFailureListener(final Object connectionID) { - connectorSet.add(entry.getConnectorPair()); + this.connectionID = connectionID; } - Iterator<Map.Entry<Pair<TransportConfiguration, TransportConfiguration>, FailoverManager>> iter = failoverManagerMap.entrySet() - .iterator(); - while (iter.hasNext()) + public void connectionFailed(final HornetQException me) { - Map.Entry<Pair<TransportConfiguration, TransportConfiguration>, FailoverManager> entry = iter.next(); + handleConnectionFailure(connectionID, me); + } + } - if (!connectorSet.contains(entry.getKey())) - { - // failoverManager no longer there - we should remove it + private static final class ActualScheduledPinger implements Runnable + { + private final WeakReference<PingRunnable> pingRunnable; - iter.remove(); - } + ActualScheduledPinger(final PingRunnable runnable) + { + pingRunnable = new WeakReference<PingRunnable>(runnable); } - for (Pair<TransportConfiguration, TransportConfiguration> connectorPair : connectorSet) + public void run() { - if (!failoverManagerMap.containsKey(connectorPair)) + PingRunnable runnable = pingRunnable.get(); + + if (runnable != null) { - // Create a new failoverManager - - FailoverManager failoverManager = new FailoverManagerImpl(this, - connectorPair.a, - connectorPair.b, - failoverOnServerShutdown, - callTimeout, - clientFailureCheckPeriod, - connectionTTL, - retryInterval, - retryIntervalMultiplier, - maxRetryInterval, - reconnectAttempts, - failoverOnInitialConnection, - threadPool, - scheduledThreadPool, - interceptors); - - failoverManagerMap.put(connectorPair, failoverManager); + runnable.run(); } } - updatefailoverManagerArray(); } - public FailoverManager[] getFailoverManagers() + private final class PingRunnable implements Runnable { - return failoverManagerArray; - } + private boolean cancelled; - // Protected ------------------------------------------------------------------------------ + private boolean first; - @Override - protected void finalize() throws Throwable - { - close(); + private long lastCheck = System.currentTimeMillis(); - super.finalize(); - } - - // Private -------------------------------------------------------------------------------- - - private void checkWrite() - { - if (readOnly) + public synchronized void run() { - throw new IllegalStateException("Cannot set attribute on SessionFactory after it has been used"); - } - } - - private ClientSession createSessionInternal(final String username, - final String password, - final boolean xa, - final boolean autoCommitSends, - final boolean autoCommitAcks, - final boolean preAcknowledge, - final int ackBatchSize) throws HornetQException - { - if (closed) - { - throw new IllegalStateException("Cannot create session, factory is closed (maybe it has been garbage collected)"); - } - - try - { - initialise(); - } - catch (Exception e) - { - throw new HornetQException(HornetQException.INTERNAL_ERROR, "Failed to initialise session factory", e); - } - - if (discoveryGroup != null && !receivedBroadcast) - { - boolean ok = discoveryGroup.waitForBroadcast(discoveryInitialWaitTimeout); - - if (!ok) + if (cancelled || stopPingingAfterOne && !first) { - throw new HornetQException(HornetQException.CONNECTION_TIMEDOUT, - "Timed out waiting to receive initial broadcast from discovery group"); + return; } - } - synchronized (this) - { - int pos = loadBalancingPolicy.select(failoverManagerArray.length); + first = false; - FailoverManager failoverManager = failoverManagerArray[pos]; + long now = System.currentTimeMillis(); - ClientSession session = failoverManager.createSession(username, - password, - xa, - autoCommitSends, - autoCommitAcks, - preAcknowledge, - ackBatchSize, - cacheLargeMessagesClient, - minLargeMessageSize, - blockOnAcknowledge, - autoGroup, - confirmationWindowSize, - producerWindowSize, - consumerWindowSize, - producerMaxRate, - consumerMaxRate, - blockOnNonDurableSend, - blockOnDurableSend, - initialMessagePacketSize, - groupID); + if (clientFailureCheckPeriod != -1 && now >= lastCheck + clientFailureCheckPeriod) + { + if (!connection.checkDataReceived()) + { + final HornetQException me = new HornetQException(HornetQException.CONNECTION_TIMEDOUT, + "Did not receive data from server for " + connection.getTransportConnection()); - return session; - } - } + cancelled = true; - private void instantiateLoadBalancingPolicy() - { - if (connectionLoadBalancingPolicyClassName == null) - { - throw new IllegalStateException("Please specify a load balancing policy class name on the session factory"); - } + threadPool.execute(new Runnable() + { + // Must be executed on different thread + public void run() + { + connection.fail(me); + } + }); - AccessController.doPrivileged(new PrivilegedAction<Object>() - { - public Object run() - { - ClassLoader loader = Thread.currentThread().getContextClassLoader(); - try - { - Class<?> clazz = loader.loadClass(connectionLoadBalancingPolicyClassName); - loadBalancingPolicy = (ConnectionLoadBalancingPolicy)clazz.newInstance(); - return null; + return; } - catch (Exception e) + else { - throw new IllegalArgumentException("Unable to instantiate load balancing policy \"" + connectionLoadBalancingPolicyClassName + - "\"", - e); + lastCheck = now; } } - }); - } - private static ClassLoader getThisClassLoader() - { - return AccessController.doPrivileged(new PrivilegedAction<ClassLoader>() - { - public ClassLoader run() - { - return ClientSessionFactoryImpl.class.getClassLoader(); - } - }); - - } + // Send a ping - private synchronized void updatefailoverManagerArray() - { - failoverManagerArray = new FailoverManager[failoverManagerMap.size()]; + Ping ping = new Ping(connectionTTL); - failoverManagerMap.values().toArray(failoverManagerArray); + Channel channel0 = connection.getChannel(0, -1); + + channel0.send(ping); + + connection.flush(); + } + + public synchronized void cancel() + { + cancelled = true; + } } - } Copied: branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl_Old.java (from rev 9287, branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java) =================================================================== --- branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl_Old.java (rev 0) +++ branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/client/impl/ClientSessionFactoryImpl_Old.java 2010-06-11 15:33:36 UTC (rev 9307) @@ -0,0 +1,1195 @@ +/* + * Copyright 2009 Red Hat, Inc. + * Red Hat 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.hornetq.core.client.impl; + +import java.io.Serializable; +import java.net.InetAddress; +import java.security.AccessController; +import java.security.PrivilegedAction; +import java.util.ArrayList; +import java.util.HashSet; +import java.util.Iterator; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.TimeUnit; + +import org.hornetq.api.core.HornetQException; +import org.hornetq.api.core.Interceptor; +import org.hornetq.api.core.Pair; +import org.hornetq.api.core.TransportConfiguration; +import org.hornetq.api.core.client.ClientSession; +import org.hornetq.api.core.client.ClientSessionFactory; +import org.hornetq.api.core.client.HornetQClient; +import org.hornetq.api.core.client.loadbalance.ConnectionLoadBalancingPolicy; +import org.hornetq.core.cluster.DiscoveryEntry; +import org.hornetq.core.cluster.DiscoveryGroup; +import org.hornetq.core.cluster.DiscoveryListener; +import org.hornetq.core.cluster.impl.DiscoveryGroupImpl; +import org.hornetq.core.logging.Logger; +import org.hornetq.utils.HornetQThreadFactory; +import org.hornetq.utils.UUIDGenerator; + +/** + * @author <a href="mailto:tim.fox@jboss.com">Tim Fox</a> + * @author <a href="mailto:clebert.suconic@jboss.org">Clebert Suconic</a> + * @author <a href="mailto:jmesnil@redhat.com">Jeff Mesnil</a> + * @author <a href="mailto:ataylor@redhat.com">Andy Taylor</a> + * @version <tt>$Revision: 3602 $</tt> + * + */ +public class ClientSessionFactoryImpl_Old implements ClientSessionFactoryInternal, DiscoveryListener, Serializable +{ + // Constants + // ------------------------------------------------------------------------------------ + + private static final long serialVersionUID = 2512460695662741413L; + + private static final Logger log = Logger.getLogger(ClientSessionFactoryImpl.class); + + // Attributes + // ----------------------------------------------------------------------------------- + + private final Map<Pair<TransportConfiguration, TransportConfiguration>, FailoverManager> failoverManagerMap = new LinkedHashMap<Pair<TransportConfiguration, TransportConfiguration>, FailoverManager>(); + + private volatile boolean receivedBroadcast = false; + + private ExecutorService threadPool; + + private ScheduledExecutorService scheduledThreadPool; + + private DiscoveryGroup discoveryGroup; + + private ConnectionLoadBalancingPolicy loadBalancingPolicy; + + private FailoverManager[] failoverManagerArray; + + private boolean readOnly; + + // Settable attributes: + + private boolean cacheLargeMessagesClient = HornetQClient.DEFAULT_CACHE_LARGE_MESSAGE_CLIENT; + + private List<Pair<TransportConfiguration, TransportConfiguration>> staticConnectors; + + private String localBindAddress; + + private String discoveryAddress; + + private int discoveryPort; + + private long discoveryRefreshTimeout; + + private long discoveryInitialWaitTimeout; + + private long clientFailureCheckPeriod; + + private long connectionTTL; + + private long callTimeout; + + private int minLargeMessageSize; + + private int consumerWindowSize; + + private int consumerMaxRate; + + private int confirmationWindowSize; + + private int producerWindowSize; + + private int producerMaxRate; + + private boolean blockOnAcknowledge; + + private boolean blockOnDurableSend; + + private boolean blockOnNonDurableSend; + + private boolean autoGroup; + + private boolean preAcknowledge; + + private String connectionLoadBalancingPolicyClassName; + + private int ackBatchSize; + + private boolean useGlobalPools; + + private int scheduledThreadPoolMaxSize; + + private int threadPoolMaxSize; + + private long retryInterval; + + private double retryIntervalMultiplier; + + private long maxRetryInterval; + + private int reconnectAttempts; + + private boolean failoverOnInitialConnection; + + private int initialMessagePacketSize; + + private volatile boolean closed; + + private boolean failoverOnServerShutdown; + + private final List<Interceptor> interceptors = new CopyOnWriteArrayList<Interceptor>(); + + private static ExecutorService globalThreadPool; + + private static ScheduledExecutorService globalScheduledThreadPool; + + private String groupID; + + private static synchronized ExecutorService getGlobalThreadPool() + { + if (ClientSessionFactoryImpl.globalThreadPool == null) + { + ThreadFactory factory = new HornetQThreadFactory("HornetQ-client-global-threads", true, getThisClassLoader()); + + ClientSessionFactoryImpl.globalThreadPool = Executors.newCachedThreadPool(factory); + } + + return ClientSessionFactoryImpl.globalThreadPool; + } + + private static synchronized ScheduledExecutorService getGlobalScheduledThreadPool() + { + if (ClientSessionFactoryImpl.globalScheduledThreadPool == null) + { + ThreadFactory factory = new HornetQThreadFactory("HornetQ-client-global-scheduled-threads", true, getThisClassLoader()); + + ClientSessionFactoryImpl.globalScheduledThreadPool = Executors.newScheduledThreadPool(HornetQClient.DEFAULT_SCHEDULED_THREAD_POOL_MAX_SIZE, + + factory); + } + + return ClientSessionFactoryImpl.globalScheduledThreadPool; + } + + private void setThreadPools() + { + if (useGlobalPools) + { + threadPool = ClientSessionFactoryImpl.getGlobalThreadPool(); + + scheduledThreadPool = ClientSessionFactoryImpl.getGlobalScheduledThreadPool(); + } + else + { + ThreadFactory factory = new HornetQThreadFactory("HornetQ-client-factory-threads-" + System.identityHashCode(this), + true, getThisClassLoader()); + + if (threadPoolMaxSize == -1) + { + threadPool = Executors.newCachedThreadPool(factory); + } + else + { + threadPool = Executors.newFixedThreadPool(threadPoolMaxSize, factory); + } + + factory = new HornetQThreadFactory("HornetQ-client-factory-pinger-threads-" + System.identityHashCode(this), + true, getThisClassLoader()); + + scheduledThreadPool = Executors.newScheduledThreadPool(scheduledThreadPoolMaxSize, factory); + } + } + + private synchronized void initialise() throws Exception + { + if (!readOnly) + { + setThreadPools(); + + instantiateLoadBalancingPolicy(); + + if (discoveryAddress != null) + { + InetAddress groupAddress = InetAddress.getByName(discoveryAddress); + + InetAddress lbAddress; + + if (localBindAddress != null) + { + lbAddress = InetAddress.getByName(localBindAddress); + } + else + { + lbAddress = null; + } + + discoveryGroup = new DiscoveryGroupImpl(UUIDGenerator.getInstance().generateStringUUID(), + discoveryAddress, + lbAddress, + groupAddress, + discoveryPort, + discoveryRefreshTimeout); + + discoveryGroup.registerListener(this); + + discoveryGroup.start(); + } + else if (staticConnectors != null) + { + for (Pair<TransportConfiguration, TransportConfiguration> pair : staticConnectors) + { + FailoverManager cm = new FailoverManagerImpl(this, + pair.a, + pair.b, + failoverOnServerShutdown, + callTimeout, + clientFailureCheckPeriod, + connectionTTL, + retryInterval, + retryIntervalMultiplier, + maxRetryInterval, + reconnectAttempts, + failoverOnInitialConnection, + threadPool, + scheduledThreadPool, + interceptors); + + failoverManagerMap.put(pair, cm); + } + + updatefailoverManagerArray(); + } + else + { + throw new IllegalStateException("Before using a session factory you must either set discovery address and port or " + "provide some static transport configuration"); + } + readOnly = true; + } + } + + // Static + // --------------------------------------------------------------------------------------- + + // Constructors + // --------------------------------------------------------------------------------- + + public ClientSessionFactoryImpl_Old(final ClientSessionFactory other) + { + localBindAddress = other.getLocalBindAddress(); + + discoveryAddress = other.getDiscoveryAddress(); + + discoveryPort = other.getDiscoveryPort(); + + staticConnectors = other.getStaticConnectors(); + + discoveryRefreshTimeout = other.getDiscoveryRefreshTimeout(); + + clientFailureCheckPeriod = other.getClientFailureCheckPeriod(); + + connectionTTL = other.getConnectionTTL(); + + callTimeout = other.getCallTimeout(); + + minLargeMessageSize = other.getMinLargeMessageSize(); + + consumerWindowSize = other.getConsumerWindowSize(); + + consumerMaxRate = other.getConsumerMaxRate(); + + confirmationWindowSize = other.getConfirmationWindowSize(); + + producerWindowSize = other.getProducerWindowSize(); + + producerMaxRate = other.getProducerMaxRate(); + + blockOnAcknowledge = other.isBlockOnAcknowledge(); + + blockOnDurableSend = other.isBlockOnDurableSend(); + + blockOnNonDurableSend = other.isBlockOnNonDurableSend(); + + autoGroup = other.isAutoGroup(); + + preAcknowledge = other.isPreAcknowledge(); + + ackBatchSize = other.getAckBatchSize(); + + connectionLoadBalancingPolicyClassName = other.getConnectionLoadBalancingPolicyClassName(); + + discoveryInitialWaitTimeout = other.getDiscoveryInitialWaitTimeout(); + + useGlobalPools = other.isUseGlobalPools(); + + scheduledThreadPoolMaxSize = other.getScheduledThreadPoolMaxSize(); + + threadPoolMaxSize = other.getThreadPoolMaxSize(); + + retryInterval = other.getRetryInterval(); + + retryIntervalMultiplier = other.getRetryIntervalMultiplier(); + + maxRetryInterval = other.getMaxRetryInterval(); + + reconnectAttempts = other.getReconnectAttempts(); + + failoverOnInitialConnection = other.isFailoverOnInitialConnection(); + + failoverOnServerShutdown = other.isFailoverOnServerShutdown(); + + cacheLargeMessagesClient = other.isCacheLargeMessagesClient(); + + initialMessagePacketSize = other.getInitialMessagePacketSize(); + + groupID = other.getGroupID(); + } + + public ClientSessionFactoryImpl_Old() + { + discoveryRefreshTimeout = HornetQClient.DEFAULT_DISCOVERY_REFRESH_TIMEOUT; + + clientFailureCheckPeriod = HornetQClient.DEFAULT_CLIENT_FAILURE_CHECK_PERIOD; + + connectionTTL = HornetQClient.DEFAULT_CONNECTION_TTL; + + callTimeout = HornetQClient.DEFAULT_CALL_TIMEOUT; + + minLargeMessageSize = HornetQClient.DEFAULT_MIN_LARGE_MESSAGE_SIZE; + + consumerWindowSize = HornetQClient.DEFAULT_CONSUMER_WINDOW_SIZE; + + consumerMaxRate = HornetQClient.DEFAULT_CONSUMER_MAX_RATE; + + confirmationWindowSize = HornetQClient.DEFAULT_CONFIRMATION_WINDOW_SIZE; + + producerWindowSize = HornetQClient.DEFAULT_PRODUCER_WINDOW_SIZE; + + producerMaxRate = HornetQClient.DEFAULT_PRODUCER_MAX_RATE; + + blockOnAcknowledge = HornetQClient.DEFAULT_BLOCK_ON_ACKNOWLEDGE; + + blockOnDurableSend = HornetQClient.DEFAULT_BLOCK_ON_DURABLE_SEND; + + blockOnNonDurableSend = HornetQClient.DEFAULT_BLOCK_ON_NON_DURABLE_SEND; + + autoGroup = HornetQClient.DEFAULT_AUTO_GROUP; + + preAcknowledge = HornetQClient.DEFAULT_PRE_ACKNOWLEDGE; + + ackBatchSize = HornetQClient.DEFAULT_ACK_BATCH_SIZE; + + connectionLoadBalancingPolicyClassName = HornetQClient.DEFAULT_CONNECTION_LOAD_BALANCING_POLICY_CLASS_NAME; + + discoveryInitialWaitTimeout = HornetQClient.DEFAULT_DISCOVERY_INITIAL_WAIT_TIMEOUT; + + useGlobalPools = HornetQClient.DEFAULT_USE_GLOBAL_POOLS; + + scheduledThreadPoolMaxSize = HornetQClient.DEFAULT_SCHEDULED_THREAD_POOL_MAX_SIZE; + + threadPoolMaxSize = HornetQClient.DEFAULT_THREAD_POOL_MAX_SIZE; + + retryInterval = HornetQClient.DEFAULT_RETRY_INTERVAL; + + retryIntervalMultiplier = HornetQClient.DEFAULT_RETRY_INTERVAL_MULTIPLIER; + + maxRetryInterval = HornetQClient.DEFAULT_MAX_RETRY_INTERVAL; + + reconnectAttempts = HornetQClient.DEFAULT_RECONNECT_ATTEMPTS; + + failoverOnInitialConnection = HornetQClient.DEFAULT_FAILOVER_ON_INITIAL_CONNECTION; + + failoverOnServerShutdown = HornetQClient.DEFAULT_FAILOVER_ON_SERVER_SHUTDOWN; + + cacheLargeMessagesClient = HornetQClient.DEFAULT_CACHE_LARGE_MESSAGE_CLIENT; + + initialMessagePacketSize = HornetQClient.DEFAULT_INITIAL_MESSAGE_PACKET_SIZE; + } + + public ClientSessionFactoryImpl_Old(final String discoveryAddress, final int discoveryPort) + { + this(); + + this.discoveryAddress = discoveryAddress; + + this.discoveryPort = discoveryPort; + } + + public ClientSessionFactoryImpl_Old(final String localBindAddress, final String discoveryAddress, final int discoveryPort) + { + this(); + + this.localBindAddress = localBindAddress; + + this.discoveryAddress = discoveryAddress; + + this.discoveryPort = discoveryPort; + } + + public ClientSessionFactoryImpl_Old(final List<Pair<TransportConfiguration, TransportConfiguration>> staticConnectors) + { + this(); + + this.staticConnectors = staticConnectors; + } + + public ClientSessionFactoryImpl_Old(final TransportConfiguration connectorConfig, + final TransportConfiguration backupConnectorConfig) + { + this(); + + staticConnectors = new ArrayList<Pair<TransportConfiguration, TransportConfiguration>>(); + + staticConnectors.add(new Pair<TransportConfiguration, TransportConfiguration>(connectorConfig, + backupConnectorConfig)); + } + + public ClientSessionFactoryImpl_Old(final TransportConfiguration connectorConfig) + { + this(connectorConfig, null); + } + + // ClientSessionFactory implementation------------------------------------------------------------ + + public synchronized boolean isCacheLargeMessagesClient() + { + return cacheLargeMessagesClient; + } + + public synchronized void setCacheLargeMessagesClient(final boolean cached) + { + cacheLargeMessagesClient = cached; + } + + public synchronized List<Pair<TransportConfiguration, TransportConfiguration>> getStaticConnectors() + { + return staticConnectors; + } + + public synchronized void setStaticConnectors(final List<Pair<TransportConfiguration, TransportConfiguration>> staticConnectors) + { + checkWrite(); + + this.staticConnectors = staticConnectors; + } + + public synchronized long getClientFailureCheckPeriod() + { + return clientFailureCheckPeriod; + } + + public synchronized void setClientFailureCheckPeriod(final long clientFailureCheckPeriod) + { + checkWrite(); + this.clientFailureCheckPeriod = clientFailureCheckPeriod; + } + + public synchronized long getConnectionTTL() + { + return connectionTTL; + } + + public synchronized void setConnectionTTL(final long connectionTTL) + { + checkWrite(); + this.connectionTTL = connectionTTL; + } + + public synchronized long getCallTimeout() + { + return callTimeout; + } + + public synchronized void setCallTimeout(final long callTimeout) + { + checkWrite(); + this.callTimeout = callTimeout; + } + + public synchronized int getMinLargeMessageSize() + { + return minLargeMessageSize; + } + + public synchronized void setMinLargeMessageSize(final int minLargeMessageSize) + { + checkWrite(); + this.minLargeMessageSize = minLargeMessageSize; + } + + public synchronized int getConsumerWindowSize() + { + return consumerWindowSize; + } + + public synchronized void setConsumerWindowSize(final int consumerWindowSize) + { + checkWrite(); + this.consumerWindowSize = consumerWindowSize; + } + + public synchronized int getConsumerMaxRate() + { + return consumerMaxRate; + } + + public synchronized void setConsumerMaxRate(final int consumerMaxRate) + { + checkWrite(); + this.consumerMaxRate = consumerMaxRate; + } + + public synchronized int getConfirmationWindowSize() + { + return confirmationWindowSize; + } + + public synchronized void setConfirmationWindowSize(final int confirmationWindowSize) + { + checkWrite(); + this.confirmationWindowSize = confirmationWindowSize; + } + + public synchronized int getProducerWindowSize() + { + return producerWindowSize; + } + + public synchronized void setProducerWindowSize(final int producerWindowSize) + { + checkWrite(); + this.producerWindowSize = producerWindowSize; + } + + public synchronized int getProducerMaxRate() + { + return producerMaxRate; + } + + public synchronized void setProducerMaxRate(final int producerMaxRate) + { + checkWrite(); + this.producerMaxRate = producerMaxRate; + } + + public synchronized boolean isBlockOnAcknowledge() + { + return blockOnAcknowledge; + } + + public synchronized void setBlockOnAcknowledge(final boolean blockOnAcknowledge) + { + checkWrite(); + this.blockOnAcknowledge = blockOnAcknowledge; + } + + public synchronized boolean isBlockOnDurableSend() + { + return blockOnDurableSend; + } + + public synchronized void setBlockOnDurableSend(final boolean blockOnDurableSend) + { + checkWrite(); + this.blockOnDurableSend = blockOnDurableSend; + } + + public synchronized boolean isBlockOnNonDurableSend() + { + return blockOnNonDurableSend; + } + + public synchronized void setBlockOnNonDurableSend(final boolean blockOnNonDurableSend) + { + checkWrite(); + this.blockOnNonDurableSend = blockOnNonDurableSend; + } + + public synchronized boolean isAutoGroup() + { + return autoGroup; + } + + public synchronized void setAutoGroup(final boolean autoGroup) + { + checkWrite(); + this.autoGroup = autoGroup; + } + + public synchronized boolean isPreAcknowledge() + { + return preAcknowledge; + } + + public synchronized void setPreAcknowledge(final boolean preAcknowledge) + { + checkWrite(); + this.preAcknowledge = preAcknowledge; + } + + public synchronized int getAckBatchSize() + { + return ackBatchSize; + } + + public synchronized void setAckBatchSize(final int ackBatchSize) + { + checkWrite(); + this.ackBatchSize = ackBatchSize; + } + + public synchronized long getDiscoveryInitialWaitTimeout() + { + return discoveryInitialWaitTimeout; + } + + public synchronized void setDiscoveryInitialWaitTimeout(final long initialWaitTimeout) + { + checkWrite(); + discoveryInitialWaitTimeout = initialWaitTimeout; + } + + public synchronized boolean isUseGlobalPools() + { + return useGlobalPools; + } + + public synchronized void setUseGlobalPools(final boolean useGlobalPools) + { + checkWrite(); + this.useGlobalPools = useGlobalPools; + } + + public synchronized int getScheduledThreadPoolMaxSize() + { + return scheduledThreadPoolMaxSize; + } + + public synchronized void setScheduledThreadPoolMaxSize(final int scheduledThreadPoolMaxSize) + { + checkWrite(); + this.scheduledThreadPoolMaxSize = scheduledThreadPoolMaxSize; + } + + public synchronized int getThreadPoolMaxSize() + { + return threadPoolMaxSize; + } + + public synchronized void setThreadPoolMaxSize(final int threadPoolMaxSize) + { + checkWrite(); + this.threadPoolMaxSize = threadPoolMaxSize; + } + + public synchronized long getRetryInterval() + { + return retryInterval; + } + + public synchronized void setRetryInterval(final long retryInterval) + { + checkWrite(); + this.retryInterval = retryInterval; + } + + public synchronized long getMaxRetryInterval() + { + return maxRetryInterval; + } + + public synchronized void setMaxRetryInterval(final long retryInterval) + { + checkWrite(); + maxRetryInterval = retryInterval; + } + + public synchronized double getRetryIntervalMultiplier() + { + return retryIntervalMultiplier; + } + + public synchronized void setRetryIntervalMultiplier(final double retryIntervalMultiplier) + { + checkWrite(); + this.retryIntervalMultiplier = retryIntervalMultiplier; + } + + public synchronized int getReconnectAttempts() + { + return reconnectAttempts; + } + + public synchronized void setReconnectAttempts(final int reconnectAttempts) + { + checkWrite(); + this.reconnectAttempts = reconnectAttempts; + } + + public synchronized boolean isFailoverOnInitialConnection() + { + return this.failoverOnInitialConnection; + } + + public synchronized void setFailoverOnInitialConnection(final boolean failover) + { + checkWrite(); + this.failoverOnInitialConnection = failover; + } + + public synchronized boolean isFailoverOnServerShutdown() + { + return failoverOnServerShutdown; + } + + public synchronized void setFailoverOnServerShutdown(final boolean failoverOnServerShutdown) + { + checkWrite(); + this.failoverOnServerShutdown = failoverOnServerShutdown; + } + + public synchronized String getConnectionLoadBalancingPolicyClassName() + { + return connectionLoadBalancingPolicyClassName; + } + + public synchronized void setConnectionLoadBalancingPolicyClassName(final String loadBalancingPolicyClassName) + { + checkWrite(); + connectionLoadBalancingPolicyClassName = loadBalancingPolicyClassName; + } + + public synchronized String getLocalBindAddress() + { + return localBindAddress; + } + + public synchronized void setLocalBindAddress(final String localBindAddress) + { + checkWrite(); + this.localBindAddress = localBindAddress; + } + + public synchronized String getDiscoveryAddress() + { + return discoveryAddress; + } + + public synchronized void setDiscoveryAddress(final String discoveryAddress) + { + checkWrite(); + this.discoveryAddress = discoveryAddress; + } + + public synchronized int getDiscoveryPort() + { + return discoveryPort; + } + + public synchronized void setDiscoveryPort(final int discoveryPort) + { + checkWrite(); + this.discoveryPort = discoveryPort; + } + + public synchronized long getDiscoveryRefreshTimeout() + { + return discoveryRefreshTimeout; + } + + public void addInterceptor(final Interceptor interceptor) + { + interceptors.add(interceptor); + } + + public boolean removeInterceptor(final Interceptor interceptor) + { + return interceptors.remove(interceptor); + } + + public synchronized void setDiscoveryRefreshTimeout(final long discoveryRefreshTimeout) + { + checkWrite(); + this.discoveryRefreshTimeout = discoveryRefreshTimeout; + } + + public synchronized int getInitialMessagePacketSize() + { + return initialMessagePacketSize; + } + + public synchronized void setInitialMessagePacketSize(final int size) + { + checkWrite(); + initialMessagePacketSize = size; + } + + public ClientSession createSession(final String username, + final String password, + final boolean xa, + final boolean autoCommitSends, + final boolean autoCommitAcks, + final boolean preAcknowledge, + final int ackBatchSize) throws HornetQException + { + return createSessionInternal(username, + password, + xa, + autoCommitSends, + autoCommitAcks, + preAcknowledge, + ackBatchSize); + } + + public ClientSession createSession(final boolean autoCommitSends, + final boolean autoCommitAcks, + final int ackBatchSize) throws HornetQException + { + return createSessionInternal(null, null, false, autoCommitSends, autoCommitAcks, preAcknowledge, ackBatchSize); + } + + public ClientSession createXASession() throws HornetQException + { + return createSessionInternal(null, null, true, false, false, preAcknowledge, ackBatchSize); + } + + public ClientSession createTransactedSession() throws HornetQException + { + return createSessionInternal(null, null, false, false, false, preAcknowledge, ackBatchSize); + } + + public ClientSession createSession() throws HornetQException + { + return createSessionInternal(null, null, false, true, true, preAcknowledge, ackBatchSize); + } + + public ClientSession createSession(final boolean autoCommitSends, final boolean autoCommitAcks) throws HornetQException + { + return createSessionInternal(null, null, false, autoCommitSends, autoCommitAcks, preAcknowledge, ackBatchSize); + } + + public ClientSession createSession(final boolean xa, final boolean autoCommitSends, final boolean autoCommitAcks) throws HornetQException + { + return createSessionInternal(null, null, xa, autoCommitSends, autoCommitAcks, preAcknowledge, ackBatchSize); + } + + public ClientSession createSession(final boolean xa, + final boolean autoCommitSends, + final boolean autoCommitAcks, + final boolean preAcknowledge) throws HornetQException + { + return createSessionInternal(null, null, xa, autoCommitSends, autoCommitAcks, preAcknowledge, ackBatchSize); + } + + public int numSessions() + { + int num = 0; + + for (FailoverManager failoverManager : failoverManagerMap.values()) + { + num += failoverManager.numSessions(); + } + + return num; + } + + public int numConnections() + { + int num = 0; + + for (FailoverManager failoverManager : failoverManagerMap.values()) + { + num += failoverManager.numConnections(); + } + + return num; + } + + public void close() + { + if (closed) + { + return; + } + + if (discoveryGroup != null) + { + try + { + discoveryGroup.stop(); + } + catch (Exception e) + { + ClientSessionFactoryImpl.log.error("Failed to stop discovery group", e); + } + } + + for (FailoverManager failoverManager : failoverManagerMap.values()) + { + failoverManager.causeExit(); + } + + failoverManagerMap.clear(); + + if (!useGlobalPools) + { + if (threadPool != null) + { + threadPool.shutdown(); + + try + { + if (!threadPool.awaitTermination(10000, TimeUnit.MILLISECONDS)) + { + ClientSessionFactoryImpl.log.warn("Timed out waiting for pool to terminate"); + } + } + catch (InterruptedException ignore) + { + } + } + + if (scheduledThreadPool != null) + { + scheduledThreadPool.shutdown(); + + try + { + if (!scheduledThreadPool.awaitTermination(10000, TimeUnit.MILLISECONDS)) + { + ClientSessionFactoryImpl.log.warn("Timed out waiting for scheduled pool to terminate"); + } + } + catch (InterruptedException ignore) + { + } + } + } + + closed = true; + } + + public ClientSessionFactory copy() + { + return new ClientSessionFactoryImpl(this); + } + + public void setGroupID(final String groupID) + { + this.groupID = groupID; + } + + public String getGroupID() + { + return groupID; + } + + // DiscoveryListener implementation -------------------------------------------------------- + + public synchronized void connectorsChanged() + { + receivedBroadcast = true; + + Map<String, DiscoveryEntry> newConnectors = discoveryGroup.getDiscoveryEntryMap(); + + Set<Pair<TransportConfiguration, TransportConfiguration>> connectorSet = new HashSet<Pair<TransportConfiguration, TransportConfiguration>>(); + + for (DiscoveryEntry entry : newConnectors.values()) + { + connectorSet.add(entry.getConnectorPair()); + } + + Iterator<Map.Entry<Pair<TransportConfiguration, TransportConfiguration>, FailoverManager>> iter = failoverManagerMap.entrySet() + .iterator(); + while (iter.hasNext()) + { + Map.Entry<Pair<TransportConfiguration, TransportConfiguration>, FailoverManager> entry = iter.next(); + + if (!connectorSet.contains(entry.getKey())) + { + // failoverManager no longer there - we should remove it + + iter.remove(); + } + } + + for (Pair<TransportConfiguration, TransportConfiguration> connectorPair : connectorSet) + { + if (!failoverManagerMap.containsKey(connectorPair)) + { + // Create a new failoverManager + + FailoverManager failoverManager = new FailoverManagerImpl(this, + connectorPair.a, + connectorPair.b, + failoverOnServerShutdown, + callTimeout, + clientFailureCheckPeriod, + connectionTTL, + retryInterval, + retryIntervalMultiplier, + maxRetryInterval, + reconnectAttempts, + failoverOnInitialConnection, + threadPool, + scheduledThreadPool, + interceptors); + + failoverManagerMap.put(connectorPair, failoverManager); + } + } + + updatefailoverManagerArray(); + } + + public FailoverManager[] getFailoverManagers() + { + return failoverManagerArray; + } + + // Protected ------------------------------------------------------------------------------ + + @Override + protected void finalize() throws Throwable + { + close(); + + super.finalize(); + } + + // Private -------------------------------------------------------------------------------- + + private void checkWrite() + { + if (readOnly) + { + throw new IllegalStateException("Cannot set attribute on SessionFactory after it has been used"); + } + } + + private ClientSession createSessionInternal(final String username, + final String password, + final boolean xa, + final boolean autoCommitSends, + final boolean autoCommitAcks, + final boolean preAcknowledge, + final int ackBatchSize) throws HornetQException + { + if (closed) + { + throw new IllegalStateException("Cannot create session, factory is closed (maybe it has been garbage collected)"); + } + + try + { + initialise(); + } + catch (Exception e) + { + throw new HornetQException(HornetQException.INTERNAL_ERROR, "Failed to initialise session factory", e); + } + + if (discoveryGroup != null && !receivedBroadcast) + { + boolean ok = discoveryGroup.waitForBroadcast(discoveryInitialWaitTimeout); + + if (!ok) + { + throw new HornetQException(HornetQException.CONNECTION_TIMEDOUT, + "Timed out waiting to receive initial broadcast from discovery group"); + } + } + + synchronized (this) + { + int pos = loadBalancingPolicy.select(failoverManagerArray.length); + + FailoverManager failoverManager = failoverManagerArray[pos]; + + ClientSession session = failoverManager.createSession(username, + password, + xa, + autoCommitSends, + autoCommitAcks, + preAcknowledge, + ackBatchSize, + cacheLargeMessagesClient, + minLargeMessageSize, + blockOnAcknowledge, + autoGroup, + confirmationWindowSize, + producerWindowSize, + consumerWindowSize, + producerMaxRate, + consumerMaxRate, + blockOnNonDurableSend, + blockOnDurableSend, + initialMessagePacketSize, + groupID); + + return session; + } + } + + private void instantiateLoadBalancingPolicy() + { + if (connectionLoadBalancingPolicyClassName == null) + { + throw new IllegalStateException("Please specify a load balancing policy class name on the session factory"); + } + + AccessController.doPrivileged(new PrivilegedAction<Object>() + { + public Object run() + { + ClassLoader loader = Thread.currentThread().getContextClassLoader(); + try + { + Class<?> clazz = loader.loadClass(connectionLoadBalancingPolicyClassName); + loadBalancingPolicy = (ConnectionLoadBalancingPolicy)clazz.newInstance(); + return null; + } + catch (Exception e) + { + throw new IllegalArgumentException("Unable to instantiate load balancing policy \"" + connectionLoadBalancingPolicyClassName + + "\"", + e); + } + } + }); + } + + private static ClassLoader getThisClassLoader() + { + return AccessController.doPrivileged(new PrivilegedAction<ClassLoader>() + { + public ClassLoader run() + { + return ClientSessionFactoryImpl.class.getClassLoader(); + } + }); + + } + + private synchronized void updatefailoverManagerArray() + { + failoverManagerArray = new FailoverManager[failoverManagerMap.size()]; + + failoverManagerMap.values().toArray(failoverManagerArray); + } + +} Modified: branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/client/impl/ClientSessionFactoryInternal.java =================================================================== --- branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/client/impl/ClientSessionFactoryInternal.java 2010-06-11 04:11:42 UTC (rev 9306) +++ branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/client/impl/ClientSessionFactoryInternal.java 2010-06-11 15:33:36 UTC (rev 9307) @@ -13,6 +13,7 @@ package org.hornetq.core.client.impl; import org.hornetq.api.core.client.ClientSessionFactory; +import org.hornetq.api.core.client.SessionFailureListener; /** * A ClientSessionFactoryInternal @@ -22,9 +23,17 @@ */ public interface ClientSessionFactoryInternal extends ClientSessionFactory { + + void addFailureListener(SessionFailureListener listener); + + boolean removeFailureListener(SessionFailureListener listener); + // for testing int numConnections(); int numSessions(); + + void removeSession(final ClientSessionInternal session); + } Modified: branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java =================================================================== --- branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java 2010-06-11 04:11:42 UTC (rev 9306) +++ branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/client/impl/ClientSessionImpl.java 2010-06-11 15:33:36 UTC (rev 9307) @@ -108,7 +108,7 @@ // Attributes ---------------------------------------------------------------------------- - private final FailoverManager failoverManager; + private final ClientSessionFactoryInternal sessionFactory; private final String name; @@ -181,10 +181,12 @@ private final String groupID; private volatile boolean inClose; + + private volatile SimpleString defaultAddress; // Constructors ---------------------------------------------------------------------------- - public ClientSessionImpl(final FailoverManager connectionManager, + public ClientSessionImpl(final ClientSessionFactoryInternal sessionFactory, final String name, final String username, final String password, @@ -211,7 +213,7 @@ final Channel channel, final Executor executor) throws HornetQException { - failoverManager = connectionManager; + this.sessionFactory = sessionFactory; this.name = name; @@ -641,15 +643,32 @@ { stop(true); } + + public void stop(final boolean waitForOnMessage) throws HornetQException + { + checkClosed(); + if (started) + { + for (ClientConsumerInternal clientConsumerInternal : consumers.values()) + { + clientConsumerInternal.stop(waitForOnMessage); + } + + channel.sendBlocking(new PacketImpl(PacketImpl.SESS_STOP)); + + started = false; + } + } + public void addFailureListener(final SessionFailureListener listener) { - failoverManager.addFailureListener(listener); + sessionFactory.addFailureListener(listener); } public boolean removeFailureListener(final SessionFailureListener listener) { - return failoverManager.removeFailureListener(listener); + return sessionFactory.removeFailureListener(listener); } public int getVersion() @@ -1028,9 +1047,12 @@ // not having any credits to send } } + + public ClientSessionFactoryInternal getSessionFactory() + { + return sessionFactory; + } - private volatile SimpleString defaultAddress; - public void setAddress(final Message message, final SimpleString address) { if (defaultAddress == null) @@ -1081,11 +1103,6 @@ channel.returnBlocking(); } - public FailoverManager getConnectionManager() - { - return failoverManager; - } - public void sendProducerCreditsMessage(final int credits, final SimpleString address) { channel.send(new SessionRequestProducerCreditsMessage(credits, address)); @@ -1275,7 +1292,7 @@ ClientSessionInternal other = (ClientSessionInternal)xares; - return failoverManager == other.getConnectionManager(); + return sessionFactory == other.getSessionFactory(); } public int prepare(final Xid xid) throws XAException @@ -1655,7 +1672,7 @@ channel.close(); } - failoverManager.removeSession(this); + sessionFactory.removeSession(this); } private void cleanUpChildren() throws Exception @@ -1700,23 +1717,8 @@ } } - public void stop(final boolean waitForOnMessage) throws HornetQException - { - checkClosed(); - if (started) - { - for (ClientConsumerInternal clientConsumerInternal : consumers.values()) - { - clientConsumerInternal.stop(waitForOnMessage); - } - channel.sendBlocking(new PacketImpl(PacketImpl.SESS_STOP)); - - started = false; - } - } - private static class BindingQueryImpl implements BindingQuery { Modified: branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/client/impl/ClientSessionInternal.java =================================================================== --- branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/client/impl/ClientSessionInternal.java 2010-06-11 04:11:42 UTC (rev 9306) +++ branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/client/impl/ClientSessionInternal.java 2010-06-11 15:33:36 UTC (rev 9307) @@ -66,7 +66,7 @@ void setForceNotSameRM(boolean force); - FailoverManager getConnectionManager(); + ClientSessionFactoryInternal getSessionFactory(); void workDone(); Modified: branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/client/impl/DelegatingSession.java =================================================================== --- branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/client/impl/DelegatingSession.java 2010-06-11 04:11:42 UTC (rev 9306) +++ branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/client/impl/DelegatingSession.java 2010-06-11 15:33:36 UTC (rev 9307) @@ -506,9 +506,9 @@ session.stop(); } - public FailoverManager getConnectionManager() + public ClientSessionFactoryInternal getSessionFactory() { - return session.getConnectionManager(); + return session.getSessionFactory(); } public void setForceNotSameRM(final boolean force) Deleted: branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/client/impl/FailoverManager.java =================================================================== --- branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/client/impl/FailoverManager.java 2010-06-11 04:11:42 UTC (rev 9306) +++ branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/client/impl/FailoverManager.java 2010-06-11 15:33:36 UTC (rev 9307) @@ -1,66 +0,0 @@ -/* - * Copyright 2009 Red Hat, Inc. - * Red Hat 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.hornetq.core.client.impl; - -import org.hornetq.api.core.HornetQException; -import org.hornetq.api.core.client.ClientSession; -import org.hornetq.api.core.client.SessionFailureListener; -import org.hornetq.core.protocol.core.CoreRemotingConnection; - -/** - * A ConnectionManager - * - * @author <a href="mailto:tim.fox@jboss.com">Tim Fox</a> - * - * Created 27 Nov 2008 18:45:46 - * - * - */ -public interface FailoverManager -{ - ClientSession createSession(final String username, - final String password, - final boolean xa, - final boolean autoCommitSends, - final boolean autoCommitAcks, - final boolean preAcknowledge, - final int ackBatchSize, - final boolean cacheLargeMessageClient, - final int minLargeMessageSize, - final boolean blockOnAcknowledge, - final boolean autoGroup, - final int confirmationWindowSize, - final int producerWindowSize, - final int consumerWindowSize, - final int producerMaxRate, - final int consumerMaxRate, - final boolean blockOnNonDurableSend, - final boolean blockOnDurableSend, - final int initialMessagePacketSize, - final String groupID) throws HornetQException; - - void removeSession(final ClientSessionInternal session); - - public CoreRemotingConnection getConnection(); - - int numConnections(); - - int numSessions(); - - void addFailureListener(SessionFailureListener listener); - - boolean removeFailureListener(SessionFailureListener listener); - - void causeExit(); -} Deleted: branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/client/impl/FailoverManagerImpl.java =================================================================== --- branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/client/impl/FailoverManagerImpl.java 2010-06-11 04:11:42 UTC (rev 9306) +++ branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/client/impl/FailoverManagerImpl.java 2010-06-11 15:33:36 UTC (rev 9307) @@ -1,1227 +0,0 @@ -/* - * Copyright 2009 Red Hat, Inc. - * Red Hat 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.hornetq.core.client.impl; - -import java.lang.ref.WeakReference; -import java.util.ArrayList; -import java.util.HashSet; -import java.util.List; -import java.util.Map; -import java.util.Set; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.Executor; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Future; -import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.locks.Lock; - -import org.hornetq.api.core.HornetQBuffer; -import org.hornetq.api.core.HornetQException; -import org.hornetq.api.core.Interceptor; -import org.hornetq.api.core.TransportConfiguration; -import org.hornetq.api.core.client.ClientSession; -import org.hornetq.api.core.client.ClientSessionFactory; -import org.hornetq.api.core.client.SessionFailureListener; -import org.hornetq.core.logging.Logger; -import org.hornetq.core.protocol.core.Channel; -import org.hornetq.core.protocol.core.ChannelHandler; -import org.hornetq.core.protocol.core.CoreRemotingConnection; -import org.hornetq.core.protocol.core.Packet; -import org.hornetq.core.protocol.core.impl.PacketImpl; -import org.hornetq.core.protocol.core.impl.RemotingConnectionImpl; -import org.hornetq.core.protocol.core.impl.wireformat.CreateSessionMessage; -import org.hornetq.core.protocol.core.impl.wireformat.CreateSessionResponseMessage; -import org.hornetq.core.protocol.core.impl.wireformat.Ping; -import org.hornetq.core.remoting.FailureListener; -import org.hornetq.core.version.Version; -import org.hornetq.spi.core.protocol.ProtocolType; -import org.hornetq.spi.core.remoting.BufferHandler; -import org.hornetq.spi.core.remoting.Connection; -import org.hornetq.spi.core.remoting.ConnectionLifeCycleListener; -import org.hornetq.spi.core.remoting.Connector; -import org.hornetq.spi.core.remoting.ConnectorFactory; -import org.hornetq.utils.ConcurrentHashSet; -import org.hornetq.utils.ConfigurationHelper; -import org.hornetq.utils.ExecutorFactory; -import org.hornetq.utils.OrderedExecutorFactory; -import org.hornetq.utils.UUIDGenerator; -import org.hornetq.utils.VersionLoader; - -/** - * A ConnectionManagerImpl - * - * @author <a href="mailto:tim.fox@jboss.com">Tim Fox</a> - * - * Created 27 Nov 2008 18:46:06 - * - */ -public class FailoverManagerImpl implements FailoverManager, ConnectionLifeCycleListener -{ - // Constants - // ------------------------------------------------------------------------------------ - - private static final long serialVersionUID = 2512460695662741413L; - - private static final Logger log = Logger.getLogger(FailoverManagerImpl.class); - - // debug - - private static Map<TransportConfiguration, Set<CoreRemotingConnection>> debugConns; - - private static boolean debug = false; - - public static void enableDebug() - { - FailoverManagerImpl.debug = true; - - FailoverManagerImpl.debugConns = new ConcurrentHashMap<TransportConfiguration, Set<CoreRemotingConnection>>(); - } - - public static void disableDebug() - { - FailoverManagerImpl.debug = false; - - FailoverManagerImpl.debugConns.clear(); - FailoverManagerImpl.debugConns = null; - } - - private void checkAddDebug(final CoreRemotingConnection conn) - { - Set<CoreRemotingConnection> conns; - - synchronized (FailoverManagerImpl.debugConns) - { - conns = FailoverManagerImpl.debugConns.get(connectorConfig); - - if (conns == null) - { - conns = new HashSet<CoreRemotingConnection>(); - - FailoverManagerImpl.debugConns.put(connectorConfig, conns); - } - - conns.add(conn); - } - } - - public static void failAllConnectionsForConnector(final TransportConfiguration config) - { - Set<CoreRemotingConnection> conns; - - synchronized (FailoverManagerImpl.debugConns) - { - conns = FailoverManagerImpl.debugConns.get(config); - - if (conns != null) - { - conns = new HashSet<CoreRemotingConnection>(FailoverManagerImpl.debugConns.get(config)); - } - } - - if (conns != null) - { - for (CoreRemotingConnection conn : conns) - { - conn.fail(new HornetQException(HornetQException.INTERNAL_ERROR, "simulated connection failure")); - } - } - } - - // Attributes - // ----------------------------------------------------------------------------------- - - private final TransportConfiguration connectorConfig; - - private ConnectorFactory connectorFactory; - - private Map<String, Object> transportParams; - - private ConnectorFactory backupConnectorFactory; - - private Map<String, Object> backupTransportParams; - - private final long callTimeout; - - private final long clientFailureCheckPeriod; - - private final long connectionTTL; - - private final Set<ClientSessionInternal> sessions = new HashSet<ClientSessionInternal>(); - - private final Object exitLock = new Object(); - - private final Object createSessionLock = new Object(); - - private boolean inCreateSession; - - private final Object failoverLock = new Object(); - - private final ExecutorFactory orderedExecutorFactory; - - private final ExecutorService threadPool; - - private final ScheduledExecutorService scheduledThreadPool; - - private final Executor closeExecutor; - - private CoreRemotingConnection connection; - - private final long retryInterval; - - private final double retryIntervalMultiplier; // For exponential backoff - - private final long maxRetryInterval; - - private final int reconnectAttempts; - - private final boolean failoverOnServerShutdown; - - private final Set<SessionFailureListener> listeners = new ConcurrentHashSet<SessionFailureListener>(); - - private Connector connector; - - private Future<?> pingerFuture; - - private PingRunnable pingRunnable; - - private volatile boolean exitLoop; - - private final List<Interceptor> interceptors; - - private volatile boolean stopPingingAfterOne; - - private final boolean failoverOnInitialConnection; - - // Static - // --------------------------------------------------------------------------------------- - - // Constructors - // --------------------------------------------------------------------------------- - - public FailoverManagerImpl(final ClientSessionFactory sessionFactory, - final TransportConfiguration connectorConfig, - final TransportConfiguration backupConfig, - final boolean failoverOnServerShutdown, - final long callTimeout, - final long clientFailureCheckPeriod, - final long connectionTTL, - final long retryInterval, - final double retryIntervalMultiplier, - final long maxRetryInterval, - final int reconnectAttempts, - final boolean failoverOnInitialConnection, - final ExecutorService threadPool, - final ScheduledExecutorService scheduledThreadPool, - final List<Interceptor> interceptors) - { - this.connectorConfig = connectorConfig; - - this.failoverOnServerShutdown = failoverOnServerShutdown; - - connectorFactory = instantiateConnectorFactory(connectorConfig.getFactoryClassName()); - - transportParams = connectorConfig.getParams(); - - checkTransportKeys(connectorFactory, transportParams); - - if (backupConfig != null) - { - backupConnectorFactory = instantiateConnectorFactory(backupConfig.getFactoryClassName()); - - backupTransportParams = backupConfig.getParams(); - - checkTransportKeys(backupConnectorFactory, backupTransportParams); - } - else - { - backupConnectorFactory = null; - - backupTransportParams = null; - } - - this.callTimeout = callTimeout; - - this.clientFailureCheckPeriod = clientFailureCheckPeriod; - - this.connectionTTL = connectionTTL; - - this.retryInterval = retryInterval; - - this.retryIntervalMultiplier = retryIntervalMultiplier; - - this.maxRetryInterval = maxRetryInterval; - - this.reconnectAttempts = reconnectAttempts; - - this.failoverOnInitialConnection = failoverOnInitialConnection; - - this.scheduledThreadPool = scheduledThreadPool; - - this.threadPool = threadPool; - - orderedExecutorFactory = new OrderedExecutorFactory(threadPool); - - closeExecutor = orderedExecutorFactory.getExecutor(); - - this.interceptors = interceptors; - } - - // ConnectionLifeCycleListener implementation -------------------------------------------------- - - public void connectionCreated(final Connection connection, final ProtocolType protocol) - { - } - - public void connectionDestroyed(final Object connectionID) - { - handleConnectionFailure(connectionID, - new HornetQException(HornetQException.NOT_CONNECTED, "Channel disconnected")); - } - - public void connectionException(final Object connectionID, final HornetQException me) - { - handleConnectionFailure(connectionID, me); - } - - // ConnectionManager implementation ------------------------------------------------------------------ - - public ClientSession createSession(final String username, - final String password, - final boolean xa, - final boolean autoCommitSends, - final boolean autoCommitAcks, - final boolean preAcknowledge, - final int ackBatchSize, - final boolean cacheLargeMessageClient, - final int minLargeMessageSize, - final boolean blockOnAcknowledge, - final boolean autoGroup, - final int confWindowSize, - final int producerWindowSize, - final int consumerWindowSize, - final int producerMaxRate, - final int consumerMaxRate, - final boolean blockOnNonDurableSend, - final boolean blockOnDurableSend, - final int initialMessagePacketSize, - final String groupID) throws HornetQException - { - synchronized (createSessionLock) - { - String name = UUIDGenerator.getInstance().generateSimpleStringUUID().toString(); - - boolean retry = false; - do - { - Version clientVersion = VersionLoader.getVersion(); - - CoreRemotingConnection theConnection = null; - - Lock lock = null; - - try - { - Channel channel1; - - synchronized (failoverLock) - { - theConnection = getConnectionWithRetry(reconnectAttempts); - - if (theConnection == null) - { - if (exitLoop) - { - return null; - } - - if (failoverOnInitialConnection && backupConnectorFactory != null) - { - // Try and connect to the backup - - log.warn("Server is not available to make initial connection to. Will " + "try backup server instead."); - - connectorFactory = backupConnectorFactory; - - transportParams = backupTransportParams; - - backupConnectorFactory = null; - - backupTransportParams = null; - - theConnection = getConnectionWithRetry(reconnectAttempts); - } - - if (exitLoop) - { - return null; - } - - if (theConnection == null) - { - throw new HornetQException(HornetQException.NOT_CONNECTED, - "Unable to connect to server using configuration " + connectorConfig); - } - } - - channel1 = theConnection.getChannel(1, -1); - - // Lock it - this must be done while the failoverLock is held - channel1.getLock().lock(); - - lock = channel1.getLock(); - } // We can now release the failoverLock - - // We now set a flag saying createSession is executing - synchronized (exitLock) - { - inCreateSession = true; - } - - long sessionChannelID = theConnection.generateChannelID(); - - Packet request = new CreateSessionMessage(name, - sessionChannelID, - clientVersion.getIncrementingVersion(), - username, - password, - minLargeMessageSize, - xa, - autoCommitSends, - autoCommitAcks, - preAcknowledge, - confWindowSize, - null); - - Packet pResponse; - try - { - pResponse = channel1.sendBlocking(request); - } - catch (HornetQException e) - { - if (e.getCode() == HornetQException.INCOMPATIBLE_CLIENT_SERVER_VERSIONS) - { - theConnection.destroy(); - } - - if (e.getCode() == HornetQException.UNBLOCKED) - { - // This means the thread was blocked on create session and failover unblocked it - // so failover could occur - - retry = true; - - continue; - } - else - { - throw e; - } - } - - CreateSessionResponseMessage response = (CreateSessionResponseMessage)pResponse; - - Channel sessionChannel = theConnection.getChannel(sessionChannelID, confWindowSize); - - ClientSessionInternal session = new ClientSessionImpl(this, - name, - username, - password, - xa, - autoCommitSends, - autoCommitAcks, - preAcknowledge, - blockOnAcknowledge, - autoGroup, - ackBatchSize, - consumerWindowSize, - consumerMaxRate, - confWindowSize, - producerWindowSize, - producerMaxRate, - blockOnNonDurableSend, - blockOnDurableSend, - cacheLargeMessageClient, - minLargeMessageSize, - initialMessagePacketSize, - groupID, - theConnection, - response.getServerVersion(), - sessionChannel, - orderedExecutorFactory.getExecutor()); - - sessions.add(session); - - ChannelHandler handler = new ClientSessionPacketHandler(session, sessionChannel); - - sessionChannel.setHandler(handler); - - return new DelegatingSession(session); - } - catch (Throwable t) - { - if (lock != null) - { - lock.unlock(); - - lock = null; - } - - if (t instanceof HornetQException) - { - throw (HornetQException)t; - } - else - { - HornetQException me = new HornetQException(HornetQException.INTERNAL_ERROR, - "Failed to create session", t); - - throw me; - } - } - finally - { - if (lock != null) - { - lock.unlock(); - } - - // Execution has finished so notify any failover thread that may be waiting for us to be done - synchronized (exitLock) - { - inCreateSession = false; - - exitLock.notify(); - } - } - } - while (retry); - } - - // Should never get here - throw new IllegalStateException("Oh my God it's full of stars!"); - } - - // Must be synchronized to prevent it happening concurrently with failover which can lead to - // inconsistencies - public void removeSession(final ClientSessionInternal session) - { - // TODO - can we simplify this locking? - synchronized (createSessionLock) - { - synchronized (failoverLock) - { - sessions.remove(session); - - checkCloseConnection(); - } - } - } - - public synchronized int numConnections() - { - return connection != null ? 1 : 0; - } - - public int numSessions() - { - return sessions.size(); - } - - public void addFailureListener(final SessionFailureListener listener) - { - listeners.add(listener); - } - - public boolean removeFailureListener(final SessionFailureListener listener) - { - return listeners.remove(listener); - } - - public void causeExit() - { - exitLoop = true; - } - - // Public - // --------------------------------------------------------------------------------------- - - public void stopPingingAfterOne() - { - stopPingingAfterOne = true; - } - - // Protected - // ------------------------------------------------------------------------------------ - - // Package Private - // ------------------------------------------------------------------------------ - - // Private - // -------------------------------------------------------------------------------------- - - private void handleConnectionFailure(final Object connectionID, final HornetQException me) - { - failoverOrReconnect(connectionID, me); - } - - private void failoverOrReconnect(final Object connectionID, final HornetQException me) - { - Set<ClientSessionInternal> sessionsToClose = null; - - synchronized (failoverLock) - { - if (connection == null || connection.getID() != connectionID) - { - // We already failed over/reconnected - probably the first failure came in, all the connections were failed - // over then a async connection exception or disconnect - // came in for one of the already exitLoop connections, so we return true - we don't want to call the - // listeners again - - return; - } - - // We call before reconnection occurs to give the user a chance to do cleanup, like cancel messages - callFailureListeners(me, false); - - // Now get locks on all channel 1s, whilst holding the failoverLock - this makes sure - // There are either no threads executing in createSession, or one is blocking on a createSession - // result. - - // Then interrupt the channel 1 that is blocking (could just interrupt them all) - - // Then release all channel 1 locks - this allows the createSession to exit the monitor - - // Then get all channel 1 locks again - this ensures the any createSession thread has executed the section and - // returned all its connections to the connection manager (the code to return connections to connection manager - // must be inside the lock - - // Then perform failover - - // Then release failoverLock - - // The other side of the bargain - during createSession: - // The calling thread must get the failoverLock and get its' connections when this is locked. - // While this is still locked it must then get the channel1 lock - // It can then release the failoverLock - // It should catch HornetQException.INTERRUPTED in the call to channel.sendBlocking - // It should then return its connections, with channel 1 lock still held - // It can then release the channel 1 lock, and retry (which will cause locking on failoverLock - // until failover is complete - - boolean serverShutdown = me.getCode() == HornetQException.DISCONNECTED; - - // We will try to failover if there is a backup connector factory, but we don't do this if the server - // has been shutdown cleanly unless failoverOnServerShutdown is true - boolean attemptFailover = backupConnectorFactory != null && (failoverOnServerShutdown || !serverShutdown); - - boolean attemptReconnect; - - if (attemptFailover) - { - attemptReconnect = false; - } - else - { - attemptReconnect = reconnectAttempts != 0; - } - - if (attemptFailover || attemptReconnect) - { - lockChannel1(); - - final boolean needToInterrupt; - - synchronized (exitLock) - { - needToInterrupt = inCreateSession; - } - - unlockChannel1(); - - if (needToInterrupt) - { - // Forcing return all channels won't guarantee that any blocked thread will return immediately - // So we need to wait for it - forceReturnChannel1(); - - // Now we need to make sure that the thread has actually exited and returned it's connections - // before failover occurs - - synchronized (exitLock) - { - while (inCreateSession) - { - try - { - exitLock.wait(5000); - } - catch (InterruptedException e) - { - } - } - } - } - - // Now we absolutely know that no threads are executing in or blocked in createSession, and no - // more will execute it until failover is complete - - // So.. do failover / reconnection - - CoreRemotingConnection oldConnection = connection; - - connection = null; - - try - { - connector.close(); - } - catch (Exception ignore) - { - } - - cancelScheduledTasks(); - - connector = null; - - if (attemptFailover) - { - // Now try failing over to backup - - connectorFactory = backupConnectorFactory; - - transportParams = backupTransportParams; - - backupConnectorFactory = null; - - backupTransportParams = null; - - reconnectSessions(oldConnection, reconnectAttempts == -1 ? -1 : reconnectAttempts + 1); - } - else - { - reconnectSessions(oldConnection, reconnectAttempts); - } - - oldConnection.destroy(); - } - else - { - connection.destroy(); - - connection = null; - } - - callFailureListeners(me, true); - - if (connection == null) - { - sessionsToClose = new HashSet<ClientSessionInternal>(sessions); - } - } - - // This needs to be outside the failover lock to prevent deadlock - if (sessionsToClose != null) - { - // If connection is null it means we didn't succeed in failing over or reconnecting - // so we close all the sessions, so they will throw exceptions when attempted to be used - - for (ClientSessionInternal session : sessionsToClose) - { - try - { - session.cleanUp(); - } - catch (Exception e) - { - FailoverManagerImpl.log.error("Failed to cleanup session"); - } - } - } - } - - private void callFailureListeners(final HornetQException me, final boolean afterReconnect) - { - final List<SessionFailureListener> listenersClone = new ArrayList<SessionFailureListener>(listeners); - - for (final SessionFailureListener listener : listenersClone) - { - try - { - if (afterReconnect) - { - listener.connectionFailed(me); - } - else - { - listener.beforeReconnect(me); - } - } - catch (final Throwable t) - { - // Failure of one listener to execute shouldn't prevent others - // from - // executing - FailoverManagerImpl.log.error("Failed to execute failure listener", t); - } - } - } - - /* - * Re-attach sessions all pre-existing sessions to the new remoting connection - */ - private void reconnectSessions(final CoreRemotingConnection oldConnection, final int reconnectAttempts) - { - CoreRemotingConnection newConnection = getConnectionWithRetry(reconnectAttempts); - - if (newConnection == null) - { - FailoverManagerImpl.log.warn("Failed to connect to server."); - - return; - } - - List<FailureListener> oldListeners = oldConnection.getFailureListeners(); - - List<FailureListener> newListeners = new ArrayList<FailureListener>(newConnection.getFailureListeners()); - - for (FailureListener listener : oldListeners) - { - // Add all apart from the first one which is the old DelegatingFailureListener - - if (listener instanceof DelegatingFailureListener == false) - { - newListeners.add(listener); - } - } - - newConnection.setFailureListeners(newListeners); - - for (ClientSessionInternal session : sessions) - { - session.handleFailover(newConnection); - } - } - - private CoreRemotingConnection getConnectionWithRetry(final int reconnectAttempts) - { - long interval = retryInterval; - - int count = 0; - - while (true) - { - if (exitLoop) - { - return null; - } - - CoreRemotingConnection theConnection = getConnection(); - - if (theConnection == null) - { - // Failed to get connection - - if (reconnectAttempts != 0) - { - count++; - - if (reconnectAttempts != -1 && count == reconnectAttempts) - { - FailoverManagerImpl.log.warn("Tried " + reconnectAttempts + " times to connect. Now giving up."); - - return null; - } - - try - { - Thread.sleep(interval); - } - catch (InterruptedException ignore) - { - } - - // Exponential back-off - long newInterval = (long)(interval * retryIntervalMultiplier); - - if (newInterval > maxRetryInterval) - { - newInterval = maxRetryInterval; - } - - interval = newInterval; - } - else - { - return null; - } - } - else - { - - if (FailoverManagerImpl.debug) - { - checkAddDebug(theConnection); - } - - return theConnection; - } - } - } - - private void cancelScheduledTasks() - { - if (pingerFuture != null) - { - pingRunnable.cancel(); - - pingerFuture.cancel(false); - - pingRunnable = null; - - pingerFuture = null; - } - } - - private void checkCloseConnection() - { - if (connection != null && sessions.size() == 0) - { - cancelScheduledTasks(); - - try - { - connection.destroy(); - } - catch (Throwable ignore) - { - } - - connection = null; - - try - { - if (connector != null) - { - connector.close(); - } - } - catch (Throwable ignore) - { - } - - connector = null; - } - } - - public CoreRemotingConnection getConnection() - { - if (connection == null) - { - Connection tc = null; - - try - { - DelegatingBufferHandler handler = new DelegatingBufferHandler(); - - connector = connectorFactory.createConnector(transportParams, - handler, - this, - closeExecutor, - threadPool, - scheduledThreadPool); - - if (connector != null) - { - connector.start(); - - tc = connector.createConnection(); - - if (tc == null) - { - try - { - connector.close(); - } - catch (Throwable t) - { - } - - connector = null; - } - } - } - catch (Exception e) - { - // Sanity catch for badly behaved remoting plugins - - FailoverManagerImpl.log.warn("connector.create or connectorFactory.createConnector should never throw an exception, implementation is badly behaved, but we'll deal with it anyway.", - e); - - if (tc != null) - { - try - { - tc.close(); - } - catch (Throwable t) - { - } - } - - if (connector != null) - { - try - { - connector.close(); - } - catch (Throwable t) - { - } - } - - tc = null; - - connector = null; - } - - if (tc == null) - { - return connection; - } - - connection = new RemotingConnectionImpl(tc, callTimeout, interceptors); - - connection.addFailureListener(new DelegatingFailureListener(connection.getID())); - - connection.getChannel(0, -1).setHandler(new Channel0Handler(connection)); - - if (clientFailureCheckPeriod != -1) - { - if (pingerFuture == null) - { - pingRunnable = new PingRunnable(); - - pingerFuture = scheduledThreadPool.scheduleWithFixedDelay(new ActualScheduledPinger(pingRunnable), - 0, - clientFailureCheckPeriod, - TimeUnit.MILLISECONDS); - } - // send a ping every time we create a new remoting connection - // to set up its TTL on the server side - else - { - pingRunnable.run(); - } - } - } - - return connection; - } - - private ConnectorFactory instantiateConnectorFactory(final String connectorFactoryClassName) - { - ClassLoader loader = Thread.currentThread().getContextClassLoader(); - try - { - Class<?> clazz = loader.loadClass(connectorFactoryClassName); - return (ConnectorFactory)clazz.newInstance(); - } - catch (Exception e) - { - throw new IllegalArgumentException("Error instantiating connector factory \"" + connectorFactoryClassName + - "\"", e); - } - } - - private void lockChannel1() - { - Channel channel1 = connection.getChannel(1, -1); - - channel1.getLock().lock(); - } - - private void unlockChannel1() - { - Channel channel1 = connection.getChannel(1, -1); - - channel1.getLock().unlock(); - } - - private void forceReturnChannel1() - { - Channel channel1 = connection.getChannel(1, -1); - - channel1.returnBlocking(); - } - - private void checkTransportKeys(final ConnectorFactory factory, final Map<String, Object> params) - { - if (params != null) - { - Set<String> invalid = ConfigurationHelper.checkKeys(factory.getAllowableProperties(), params.keySet()); - - if (!invalid.isEmpty()) - { - String msg = ConfigurationHelper.stringSetToCommaListString("The following keys are invalid for configuring a connector: ", - invalid); - - throw new IllegalStateException(msg); - - } - } - } - - private class Channel0Handler implements ChannelHandler - { - private final CoreRemotingConnection conn; - - private Channel0Handler(final CoreRemotingConnection conn) - { - this.conn = conn; - } - - public void handlePacket(final Packet packet) - { - final byte type = packet.getType(); - - if (type == PacketImpl.DISCONNECT) - { - closeExecutor.execute(new Runnable() - { - // Must be executed on new thread since cannot block the netty thread for a long time and fail can - // cause reconnect loop - public void run() - { - conn.fail(new HornetQException(HornetQException.DISCONNECTED, - "The connection was disconnected because of server shutdown")); - } - }); - } - } - } - - private class DelegatingBufferHandler implements BufferHandler - { - public void bufferReceived(final Object connectionID, final HornetQBuffer buffer) - { - CoreRemotingConnection theConn = connection; - - if (theConn != null && connectionID == theConn.getID()) - { - theConn.bufferReceived(connectionID, buffer); - } - } - } - - private class DelegatingFailureListener implements FailureListener - { - private final Object connectionID; - - DelegatingFailureListener(final Object connectionID) - { - this.connectionID = connectionID; - } - - public void connectionFailed(final HornetQException me) - { - handleConnectionFailure(connectionID, me); - } - } - - private static final class ActualScheduledPinger implements Runnable - { - private final WeakReference<PingRunnable> pingRunnable; - - ActualScheduledPinger(final PingRunnable runnable) - { - pingRunnable = new WeakReference<PingRunnable>(runnable); - } - - public void run() - { - PingRunnable runnable = pingRunnable.get(); - - if (runnable != null) - { - runnable.run(); - } - } - - } - - private final class PingRunnable implements Runnable - { - private boolean cancelled; - - private boolean first; - - private long lastCheck = System.currentTimeMillis(); - - public synchronized void run() - { - if (cancelled || stopPingingAfterOne && !first) - { - return; - } - - first = false; - - long now = System.currentTimeMillis(); - - if (clientFailureCheckPeriod != -1 && now >= lastCheck + clientFailureCheckPeriod) - { - if (!connection.checkDataReceived()) - { - final HornetQException me = new HornetQException(HornetQException.CONNECTION_TIMEDOUT, - "Did not receive data from server for " + connection.getTransportConnection()); - - cancelled = true; - - threadPool.execute(new Runnable() - { - // Must be executed on different thread - public void run() - { - connection.fail(me); - } - }); - - return; - } - else - { - lastCheck = now; - } - } - - // Send a ping - - Ping ping = new Ping(connectionTTL); - - Channel channel0 = connection.getChannel(0, -1); - - channel0.send(ping); - - connection.flush(); - } - - public synchronized void cancel() - { - cancelled = true; - } - } - -} Copied: branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/client/impl/FailoverManagerImpl_Old.java (from rev 9287, branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/client/impl/FailoverManagerImpl.java) =================================================================== --- branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/client/impl/FailoverManagerImpl_Old.java (rev 0) +++ branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/client/impl/FailoverManagerImpl_Old.java 2010-06-11 15:33:36 UTC (rev 9307) @@ -0,0 +1,1227 @@ +/* + * Copyright 2009 Red Hat, Inc. + * Red Hat 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.hornetq.core.client.impl; + +import java.lang.ref.WeakReference; +import java.util.ArrayList; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Future; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.locks.Lock; + +import org.hornetq.api.core.HornetQBuffer; +import org.hornetq.api.core.HornetQException; +import org.hornetq.api.core.Interceptor; +import org.hornetq.api.core.TransportConfiguration; +import org.hornetq.api.core.client.ClientSession; +import org.hornetq.api.core.client.ClientSessionFactory; +import org.hornetq.api.core.client.SessionFailureListener; +import org.hornetq.core.logging.Logger; +import org.hornetq.core.protocol.core.Channel; +import org.hornetq.core.protocol.core.ChannelHandler; +import org.hornetq.core.protocol.core.CoreRemotingConnection; +import org.hornetq.core.protocol.core.Packet; +import org.hornetq.core.protocol.core.impl.PacketImpl; +import org.hornetq.core.protocol.core.impl.RemotingConnectionImpl; +import org.hornetq.core.protocol.core.impl.wireformat.CreateSessionMessage; +import org.hornetq.core.protocol.core.impl.wireformat.CreateSessionResponseMessage; +import org.hornetq.core.protocol.core.impl.wireformat.Ping; +import org.hornetq.core.remoting.FailureListener; +import org.hornetq.core.version.Version; +import org.hornetq.spi.core.protocol.ProtocolType; +import org.hornetq.spi.core.remoting.BufferHandler; +import org.hornetq.spi.core.remoting.Connection; +import org.hornetq.spi.core.remoting.ConnectionLifeCycleListener; +import org.hornetq.spi.core.remoting.Connector; +import org.hornetq.spi.core.remoting.ConnectorFactory; +import org.hornetq.utils.ConcurrentHashSet; +import org.hornetq.utils.ConfigurationHelper; +import org.hornetq.utils.ExecutorFactory; +import org.hornetq.utils.OrderedExecutorFactory; +import org.hornetq.utils.UUIDGenerator; +import org.hornetq.utils.VersionLoader; + +/** + * A ConnectionManagerImpl + * + * @author <a href="mailto:tim.fox@jboss.com">Tim Fox</a> + * + * Created 27 Nov 2008 18:46:06 + * + */ +public class FailoverManagerImpl_Old implements FailoverManager, ConnectionLifeCycleListener +{ + // Constants + // ------------------------------------------------------------------------------------ + + private static final long serialVersionUID = 2512460695662741413L; + + private static final Logger log = Logger.getLogger(FailoverManagerImpl.class); + + // debug + + private static Map<TransportConfiguration, Set<CoreRemotingConnection>> debugConns; + + private static boolean debug = false; + + public static void enableDebug() + { + FailoverManagerImpl.debug = true; + + FailoverManagerImpl.debugConns = new ConcurrentHashMap<TransportConfiguration, Set<CoreRemotingConnection>>(); + } + + public static void disableDebug() + { + FailoverManagerImpl.debug = false; + + FailoverManagerImpl.debugConns.clear(); + FailoverManagerImpl.debugConns = null; + } + + private void checkAddDebug(final CoreRemotingConnection conn) + { + Set<CoreRemotingConnection> conns; + + synchronized (FailoverManagerImpl.debugConns) + { + conns = FailoverManagerImpl.debugConns.get(connectorConfig); + + if (conns == null) + { + conns = new HashSet<CoreRemotingConnection>(); + + FailoverManagerImpl.debugConns.put(connectorConfig, conns); + } + + conns.add(conn); + } + } + + public static void failAllConnectionsForConnector(final TransportConfiguration config) + { + Set<CoreRemotingConnection> conns; + + synchronized (FailoverManagerImpl.debugConns) + { + conns = FailoverManagerImpl.debugConns.get(config); + + if (conns != null) + { + conns = new HashSet<CoreRemotingConnection>(FailoverManagerImpl.debugConns.get(config)); + } + } + + if (conns != null) + { + for (CoreRemotingConnection conn : conns) + { + conn.fail(new HornetQException(HornetQException.INTERNAL_ERROR, "simulated connection failure")); + } + } + } + + // Attributes + // ----------------------------------------------------------------------------------- + + private final TransportConfiguration connectorConfig; + + private ConnectorFactory connectorFactory; + + private Map<String, Object> transportParams; + + private ConnectorFactory backupConnectorFactory; + + private Map<String, Object> backupTransportParams; + + private final long callTimeout; + + private final long clientFailureCheckPeriod; + + private final long connectionTTL; + + private final Set<ClientSessionInternal> sessions = new HashSet<ClientSessionInternal>(); + + private final Object exitLock = new Object(); + + private final Object createSessionLock = new Object(); + + private boolean inCreateSession; + + private final Object failoverLock = new Object(); + + private final ExecutorFactory orderedExecutorFactory; + + private final ExecutorService threadPool; + + private final ScheduledExecutorService scheduledThreadPool; + + private final Executor closeExecutor; + + private CoreRemotingConnection connection; + + private final long retryInterval; + + private final double retryIntervalMultiplier; // For exponential backoff + + private final long maxRetryInterval; + + private final int reconnectAttempts; + + private final boolean failoverOnServerShutdown; + + private final Set<SessionFailureListener> listeners = new ConcurrentHashSet<SessionFailureListener>(); + + private Connector connector; + + private Future<?> pingerFuture; + + private PingRunnable pingRunnable; + + private volatile boolean exitLoop; + + private final List<Interceptor> interceptors; + + private volatile boolean stopPingingAfterOne; + + private final boolean failoverOnInitialConnection; + + // Static + // --------------------------------------------------------------------------------------- + + // Constructors + // --------------------------------------------------------------------------------- + + public FailoverManagerImpl_Old(final ClientSessionFactory sessionFactory, + final TransportConfiguration connectorConfig, + final TransportConfiguration backupConfig, + final boolean failoverOnServerShutdown, + final long callTimeout, + final long clientFailureCheckPeriod, + final long connectionTTL, + final long retryInterval, + final double retryIntervalMultiplier, + final long maxRetryInterval, + final int reconnectAttempts, + final boolean failoverOnInitialConnection, + final ExecutorService threadPool, + final ScheduledExecutorService scheduledThreadPool, + final List<Interceptor> interceptors) + { + this.connectorConfig = connectorConfig; + + this.failoverOnServerShutdown = failoverOnServerShutdown; + + connectorFactory = instantiateConnectorFactory(connectorConfig.getFactoryClassName()); + + transportParams = connectorConfig.getParams(); + + checkTransportKeys(connectorFactory, transportParams); + + if (backupConfig != null) + { + backupConnectorFactory = instantiateConnectorFactory(backupConfig.getFactoryClassName()); + + backupTransportParams = backupConfig.getParams(); + + checkTransportKeys(backupConnectorFactory, backupTransportParams); + } + else + { + backupConnectorFactory = null; + + backupTransportParams = null; + } + + this.callTimeout = callTimeout; + + this.clientFailureCheckPeriod = clientFailureCheckPeriod; + + this.connectionTTL = connectionTTL; + + this.retryInterval = retryInterval; + + this.retryIntervalMultiplier = retryIntervalMultiplier; + + this.maxRetryInterval = maxRetryInterval; + + this.reconnectAttempts = reconnectAttempts; + + this.failoverOnInitialConnection = failoverOnInitialConnection; + + this.scheduledThreadPool = scheduledThreadPool; + + this.threadPool = threadPool; + + orderedExecutorFactory = new OrderedExecutorFactory(threadPool); + + closeExecutor = orderedExecutorFactory.getExecutor(); + + this.interceptors = interceptors; + } + + // ConnectionLifeCycleListener implementation -------------------------------------------------- + + public void connectionCreated(final Connection connection, final ProtocolType protocol) + { + } + + public void connectionDestroyed(final Object connectionID) + { + handleConnectionFailure(connectionID, + new HornetQException(HornetQException.NOT_CONNECTED, "Channel disconnected")); + } + + public void connectionException(final Object connectionID, final HornetQException me) + { + handleConnectionFailure(connectionID, me); + } + + // ConnectionManager implementation ------------------------------------------------------------------ + + public ClientSession createSession(final String username, + final String password, + final boolean xa, + final boolean autoCommitSends, + final boolean autoCommitAcks, + final boolean preAcknowledge, + final int ackBatchSize, + final boolean cacheLargeMessageClient, + final int minLargeMessageSize, + final boolean blockOnAcknowledge, + final boolean autoGroup, + final int confWindowSize, + final int producerWindowSize, + final int consumerWindowSize, + final int producerMaxRate, + final int consumerMaxRate, + final boolean blockOnNonDurableSend, + final boolean blockOnDurableSend, + final int initialMessagePacketSize, + final String groupID) throws HornetQException + { + synchronized (createSessionLock) + { + String name = UUIDGenerator.getInstance().generateSimpleStringUUID().toString(); + + boolean retry = false; + do + { + Version clientVersion = VersionLoader.getVersion(); + + CoreRemotingConnection theConnection = null; + + Lock lock = null; + + try + { + Channel channel1; + + synchronized (failoverLock) + { + theConnection = getConnectionWithRetry(reconnectAttempts); + + if (theConnection == null) + { + if (exitLoop) + { + return null; + } + + if (failoverOnInitialConnection && backupConnectorFactory != null) + { + // Try and connect to the backup + + log.warn("Server is not available to make initial connection to. Will " + "try backup server instead."); + + connectorFactory = backupConnectorFactory; + + transportParams = backupTransportParams; + + backupConnectorFactory = null; + + backupTransportParams = null; + + theConnection = getConnectionWithRetry(reconnectAttempts); + } + + if (exitLoop) + { + return null; + } + + if (theConnection == null) + { + throw new HornetQException(HornetQException.NOT_CONNECTED, + "Unable to connect to server using configuration " + connectorConfig); + } + } + + channel1 = theConnection.getChannel(1, -1); + + // Lock it - this must be done while the failoverLock is held + channel1.getLock().lock(); + + lock = channel1.getLock(); + } // We can now release the failoverLock + + // We now set a flag saying createSession is executing + synchronized (exitLock) + { + inCreateSession = true; + } + + long sessionChannelID = theConnection.generateChannelID(); + + Packet request = new CreateSessionMessage(name, + sessionChannelID, + clientVersion.getIncrementingVersion(), + username, + password, + minLargeMessageSize, + xa, + autoCommitSends, + autoCommitAcks, + preAcknowledge, + confWindowSize, + null); + + Packet pResponse; + try + { + pResponse = channel1.sendBlocking(request); + } + catch (HornetQException e) + { + if (e.getCode() == HornetQException.INCOMPATIBLE_CLIENT_SERVER_VERSIONS) + { + theConnection.destroy(); + } + + if (e.getCode() == HornetQException.UNBLOCKED) + { + // This means the thread was blocked on create session and failover unblocked it + // so failover could occur + + retry = true; + + continue; + } + else + { + throw e; + } + } + + CreateSessionResponseMessage response = (CreateSessionResponseMessage)pResponse; + + Channel sessionChannel = theConnection.getChannel(sessionChannelID, confWindowSize); + + ClientSessionInternal session = new ClientSessionImpl(this, + name, + username, + password, + xa, + autoCommitSends, + autoCommitAcks, + preAcknowledge, + blockOnAcknowledge, + autoGroup, + ackBatchSize, + consumerWindowSize, + consumerMaxRate, + confWindowSize, + producerWindowSize, + producerMaxRate, + blockOnNonDurableSend, + blockOnDurableSend, + cacheLargeMessageClient, + minLargeMessageSize, + initialMessagePacketSize, + groupID, + theConnection, + response.getServerVersion(), + sessionChannel, + orderedExecutorFactory.getExecutor()); + + sessions.add(session); + + ChannelHandler handler = new ClientSessionPacketHandler(session, sessionChannel); + + sessionChannel.setHandler(handler); + + return new DelegatingSession(session); + } + catch (Throwable t) + { + if (lock != null) + { + lock.unlock(); + + lock = null; + } + + if (t instanceof HornetQException) + { + throw (HornetQException)t; + } + else + { + HornetQException me = new HornetQException(HornetQException.INTERNAL_ERROR, + "Failed to create session", t); + + throw me; + } + } + finally + { + if (lock != null) + { + lock.unlock(); + } + + // Execution has finished so notify any failover thread that may be waiting for us to be done + synchronized (exitLock) + { + inCreateSession = false; + + exitLock.notify(); + } + } + } + while (retry); + } + + // Should never get here + throw new IllegalStateException("Oh my God it's full of stars!"); + } + + // Must be synchronized to prevent it happening concurrently with failover which can lead to + // inconsistencies + public void removeSession(final ClientSessionInternal session) + { + // TODO - can we simplify this locking? + synchronized (createSessionLock) + { + synchronized (failoverLock) + { + sessions.remove(session); + + checkCloseConnection(); + } + } + } + + public synchronized int numConnections() + { + return connection != null ? 1 : 0; + } + + public int numSessions() + { + return sessions.size(); + } + + public void addFailureListener(final SessionFailureListener listener) + { + listeners.add(listener); + } + + public boolean removeFailureListener(final SessionFailureListener listener) + { + return listeners.remove(listener); + } + + public void causeExit() + { + exitLoop = true; + } + + // Public + // --------------------------------------------------------------------------------------- + + public void stopPingingAfterOne() + { + stopPingingAfterOne = true; + } + + // Protected + // ------------------------------------------------------------------------------------ + + // Package Private + // ------------------------------------------------------------------------------ + + // Private + // -------------------------------------------------------------------------------------- + + private void handleConnectionFailure(final Object connectionID, final HornetQException me) + { + failoverOrReconnect(connectionID, me); + } + + private void failoverOrReconnect(final Object connectionID, final HornetQException me) + { + Set<ClientSessionInternal> sessionsToClose = null; + + synchronized (failoverLock) + { + if (connection == null || connection.getID() != connectionID) + { + // We already failed over/reconnected - probably the first failure came in, all the connections were failed + // over then a async connection exception or disconnect + // came in for one of the already exitLoop connections, so we return true - we don't want to call the + // listeners again + + return; + } + + // We call before reconnection occurs to give the user a chance to do cleanup, like cancel messages + callFailureListeners(me, false); + + // Now get locks on all channel 1s, whilst holding the failoverLock - this makes sure + // There are either no threads executing in createSession, or one is blocking on a createSession + // result. + + // Then interrupt the channel 1 that is blocking (could just interrupt them all) + + // Then release all channel 1 locks - this allows the createSession to exit the monitor + + // Then get all channel 1 locks again - this ensures the any createSession thread has executed the section and + // returned all its connections to the connection manager (the code to return connections to connection manager + // must be inside the lock + + // Then perform failover + + // Then release failoverLock + + // The other side of the bargain - during createSession: + // The calling thread must get the failoverLock and get its' connections when this is locked. + // While this is still locked it must then get the channel1 lock + // It can then release the failoverLock + // It should catch HornetQException.INTERRUPTED in the call to channel.sendBlocking + // It should then return its connections, with channel 1 lock still held + // It can then release the channel 1 lock, and retry (which will cause locking on failoverLock + // until failover is complete + + boolean serverShutdown = me.getCode() == HornetQException.DISCONNECTED; + + // We will try to failover if there is a backup connector factory, but we don't do this if the server + // has been shutdown cleanly unless failoverOnServerShutdown is true + boolean attemptFailover = backupConnectorFactory != null && (failoverOnServerShutdown || !serverShutdown); + + boolean attemptReconnect; + + if (attemptFailover) + { + attemptReconnect = false; + } + else + { + attemptReconnect = reconnectAttempts != 0; + } + + if (attemptFailover || attemptReconnect) + { + lockChannel1(); + + final boolean needToInterrupt; + + synchronized (exitLock) + { + needToInterrupt = inCreateSession; + } + + unlockChannel1(); + + if (needToInterrupt) + { + // Forcing return all channels won't guarantee that any blocked thread will return immediately + // So we need to wait for it + forceReturnChannel1(); + + // Now we need to make sure that the thread has actually exited and returned it's connections + // before failover occurs + + synchronized (exitLock) + { + while (inCreateSession) + { + try + { + exitLock.wait(5000); + } + catch (InterruptedException e) + { + } + } + } + } + + // Now we absolutely know that no threads are executing in or blocked in createSession, and no + // more will execute it until failover is complete + + // So.. do failover / reconnection + + CoreRemotingConnection oldConnection = connection; + + connection = null; + + try + { + connector.close(); + } + catch (Exception ignore) + { + } + + cancelScheduledTasks(); + + connector = null; + + if (attemptFailover) + { + // Now try failing over to backup + + connectorFactory = backupConnectorFactory; + + transportParams = backupTransportParams; + + backupConnectorFactory = null; + + backupTransportParams = null; + + reconnectSessions(oldConnection, reconnectAttempts == -1 ? -1 : reconnectAttempts + 1); + } + else + { + reconnectSessions(oldConnection, reconnectAttempts); + } + + oldConnection.destroy(); + } + else + { + connection.destroy(); + + connection = null; + } + + callFailureListeners(me, true); + + if (connection == null) + { + sessionsToClose = new HashSet<ClientSessionInternal>(sessions); + } + } + + // This needs to be outside the failover lock to prevent deadlock + if (sessionsToClose != null) + { + // If connection is null it means we didn't succeed in failing over or reconnecting + // so we close all the sessions, so they will throw exceptions when attempted to be used + + for (ClientSessionInternal session : sessionsToClose) + { + try + { + session.cleanUp(); + } + catch (Exception e) + { + FailoverManagerImpl.log.error("Failed to cleanup session"); + } + } + } + } + + private void callFailureListeners(final HornetQException me, final boolean afterReconnect) + { + final List<SessionFailureListener> listenersClone = new ArrayList<SessionFailureListener>(listeners); + + for (final SessionFailureListener listener : listenersClone) + { + try + { + if (afterReconnect) + { + listener.connectionFailed(me); + } + else + { + listener.beforeReconnect(me); + } + } + catch (final Throwable t) + { + // Failure of one listener to execute shouldn't prevent others + // from + // executing + FailoverManagerImpl.log.error("Failed to execute failure listener", t); + } + } + } + + /* + * Re-attach sessions all pre-existing sessions to the new remoting connection + */ + private void reconnectSessions(final CoreRemotingConnection oldConnection, final int reconnectAttempts) + { + CoreRemotingConnection newConnection = getConnectionWithRetry(reconnectAttempts); + + if (newConnection == null) + { + FailoverManagerImpl.log.warn("Failed to connect to server."); + + return; + } + + List<FailureListener> oldListeners = oldConnection.getFailureListeners(); + + List<FailureListener> newListeners = new ArrayList<FailureListener>(newConnection.getFailureListeners()); + + for (FailureListener listener : oldListeners) + { + // Add all apart from the first one which is the old DelegatingFailureListener + + if (listener instanceof DelegatingFailureListener == false) + { + newListeners.add(listener); + } + } + + newConnection.setFailureListeners(newListeners); + + for (ClientSessionInternal session : sessions) + { + session.handleFailover(newConnection); + } + } + + private CoreRemotingConnection getConnectionWithRetry(final int reconnectAttempts) + { + long interval = retryInterval; + + int count = 0; + + while (true) + { + if (exitLoop) + { + return null; + } + + CoreRemotingConnection theConnection = getConnection(); + + if (theConnection == null) + { + // Failed to get connection + + if (reconnectAttempts != 0) + { + count++; + + if (reconnectAttempts != -1 && count == reconnectAttempts) + { + FailoverManagerImpl.log.warn("Tried " + reconnectAttempts + " times to connect. Now giving up."); + + return null; + } + + try + { + Thread.sleep(interval); + } + catch (InterruptedException ignore) + { + } + + // Exponential back-off + long newInterval = (long)(interval * retryIntervalMultiplier); + + if (newInterval > maxRetryInterval) + { + newInterval = maxRetryInterval; + } + + interval = newInterval; + } + else + { + return null; + } + } + else + { + + if (FailoverManagerImpl.debug) + { + checkAddDebug(theConnection); + } + + return theConnection; + } + } + } + + private void cancelScheduledTasks() + { + if (pingerFuture != null) + { + pingRunnable.cancel(); + + pingerFuture.cancel(false); + + pingRunnable = null; + + pingerFuture = null; + } + } + + private void checkCloseConnection() + { + if (connection != null && sessions.size() == 0) + { + cancelScheduledTasks(); + + try + { + connection.destroy(); + } + catch (Throwable ignore) + { + } + + connection = null; + + try + { + if (connector != null) + { + connector.close(); + } + } + catch (Throwable ignore) + { + } + + connector = null; + } + } + + public CoreRemotingConnection getConnection() + { + if (connection == null) + { + Connection tc = null; + + try + { + DelegatingBufferHandler handler = new DelegatingBufferHandler(); + + connector = connectorFactory.createConnector(transportParams, + handler, + this, + closeExecutor, + threadPool, + scheduledThreadPool); + + if (connector != null) + { + connector.start(); + + tc = connector.createConnection(); + + if (tc == null) + { + try + { + connector.close(); + } + catch (Throwable t) + { + } + + connector = null; + } + } + } + catch (Exception e) + { + // Sanity catch for badly behaved remoting plugins + + FailoverManagerImpl.log.warn("connector.create or connectorFactory.createConnector should never throw an exception, implementation is badly behaved, but we'll deal with it anyway.", + e); + + if (tc != null) + { + try + { + tc.close(); + } + catch (Throwable t) + { + } + } + + if (connector != null) + { + try + { + connector.close(); + } + catch (Throwable t) + { + } + } + + tc = null; + + connector = null; + } + + if (tc == null) + { + return connection; + } + + connection = new RemotingConnectionImpl(tc, callTimeout, interceptors); + + connection.addFailureListener(new DelegatingFailureListener(connection.getID())); + + connection.getChannel(0, -1).setHandler(new Channel0Handler(connection)); + + if (clientFailureCheckPeriod != -1) + { + if (pingerFuture == null) + { + pingRunnable = new PingRunnable(); + + pingerFuture = scheduledThreadPool.scheduleWithFixedDelay(new ActualScheduledPinger(pingRunnable), + 0, + clientFailureCheckPeriod, + TimeUnit.MILLISECONDS); + } + // send a ping every time we create a new remoting connection + // to set up its TTL on the server side + else + { + pingRunnable.run(); + } + } + } + + return connection; + } + + private ConnectorFactory instantiateConnectorFactory(final String connectorFactoryClassName) + { + ClassLoader loader = Thread.currentThread().getContextClassLoader(); + try + { + Class<?> clazz = loader.loadClass(connectorFactoryClassName); + return (ConnectorFactory)clazz.newInstance(); + } + catch (Exception e) + { + throw new IllegalArgumentException("Error instantiating connector factory \"" + connectorFactoryClassName + + "\"", e); + } + } + + private void lockChannel1() + { + Channel channel1 = connection.getChannel(1, -1); + + channel1.getLock().lock(); + } + + private void unlockChannel1() + { + Channel channel1 = connection.getChannel(1, -1); + + channel1.getLock().unlock(); + } + + private void forceReturnChannel1() + { + Channel channel1 = connection.getChannel(1, -1); + + channel1.returnBlocking(); + } + + private void checkTransportKeys(final ConnectorFactory factory, final Map<String, Object> params) + { + if (params != null) + { + Set<String> invalid = ConfigurationHelper.checkKeys(factory.getAllowableProperties(), params.keySet()); + + if (!invalid.isEmpty()) + { + String msg = ConfigurationHelper.stringSetToCommaListString("The following keys are invalid for configuring a connector: ", + invalid); + + throw new IllegalStateException(msg); + + } + } + } + + private class Channel0Handler implements ChannelHandler + { + private final CoreRemotingConnection conn; + + private Channel0Handler(final CoreRemotingConnection conn) + { + this.conn = conn; + } + + public void handlePacket(final Packet packet) + { + final byte type = packet.getType(); + + if (type == PacketImpl.DISCONNECT) + { + closeExecutor.execute(new Runnable() + { + // Must be executed on new thread since cannot block the netty thread for a long time and fail can + // cause reconnect loop + public void run() + { + conn.fail(new HornetQException(HornetQException.DISCONNECTED, + "The connection was disconnected because of server shutdown")); + } + }); + } + } + } + + private class DelegatingBufferHandler implements BufferHandler + { + public void bufferReceived(final Object connectionID, final HornetQBuffer buffer) + { + CoreRemotingConnection theConn = connection; + + if (theConn != null && connectionID == theConn.getID()) + { + theConn.bufferReceived(connectionID, buffer); + } + } + } + + private class DelegatingFailureListener implements FailureListener + { + private final Object connectionID; + + DelegatingFailureListener(final Object connectionID) + { + this.connectionID = connectionID; + } + + public void connectionFailed(final HornetQException me) + { + handleConnectionFailure(connectionID, me); + } + } + + private static final class ActualScheduledPinger implements Runnable + { + private final WeakReference<PingRunnable> pingRunnable; + + ActualScheduledPinger(final PingRunnable runnable) + { + pingRunnable = new WeakReference<PingRunnable>(runnable); + } + + public void run() + { + PingRunnable runnable = pingRunnable.get(); + + if (runnable != null) + { + runnable.run(); + } + } + + } + + private final class PingRunnable implements Runnable + { + private boolean cancelled; + + private boolean first; + + private long lastCheck = System.currentTimeMillis(); + + public synchronized void run() + { + if (cancelled || stopPingingAfterOne && !first) + { + return; + } + + first = false; + + long now = System.currentTimeMillis(); + + if (clientFailureCheckPeriod != -1 && now >= lastCheck + clientFailureCheckPeriod) + { + if (!connection.checkDataReceived()) + { + final HornetQException me = new HornetQException(HornetQException.CONNECTION_TIMEDOUT, + "Did not receive data from server for " + connection.getTransportConnection()); + + cancelled = true; + + threadPool.execute(new Runnable() + { + // Must be executed on different thread + public void run() + { + connection.fail(me); + } + }); + + return; + } + else + { + lastCheck = now; + } + } + + // Send a ping + + Ping ping = new Ping(connectionTTL); + + Channel channel0 = connection.getChannel(0, -1); + + channel0.send(ping); + + connection.flush(); + } + + public synchronized void cancel() + { + cancelled = true; + } + } + +} Copied: branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/client/impl/FailoverManager_Old.java (from rev 9287, branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/client/impl/FailoverManager.java) =================================================================== --- branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/client/impl/FailoverManager_Old.java (rev 0) +++ branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/client/impl/FailoverManager_Old.java 2010-06-11 15:33:36 UTC (rev 9307) @@ -0,0 +1,66 @@ +/* + * Copyright 2009 Red Hat, Inc. + * Red Hat 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.hornetq.core.client.impl; + +import org.hornetq.api.core.HornetQException; +import org.hornetq.api.core.client.ClientSession; +import org.hornetq.api.core.client.SessionFailureListener; +import org.hornetq.core.protocol.core.CoreRemotingConnection; + +/** + * A ConnectionManager + * + * @author <a href="mailto:tim.fox@jboss.com">Tim Fox</a> + * + * Created 27 Nov 2008 18:45:46 + * + * + */ +public interface FailoverManager_Old +{ + ClientSession createSession(final String username, + final String password, + final boolean xa, + final boolean autoCommitSends, + final boolean autoCommitAcks, + final boolean preAcknowledge, + final int ackBatchSize, + final boolean cacheLargeMessageClient, + final int minLargeMessageSize, + final boolean blockOnAcknowledge, + final boolean autoGroup, + final int confirmationWindowSize, + final int producerWindowSize, + final int consumerWindowSize, + final int producerMaxRate, + final int consumerMaxRate, + final boolean blockOnNonDurableSend, + final boolean blockOnDurableSend, + final int initialMessagePacketSize, + final String groupID) throws HornetQException; + + void removeSession(final ClientSessionInternal session); + + public CoreRemotingConnection getConnection(); + + int numConnections(); + + int numSessions(); + + void addFailureListener(SessionFailureListener listener); + + boolean removeFailureListener(SessionFailureListener listener); + + void causeExit(); +} Modified: branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/cluster/DiscoveryEntry.java =================================================================== --- branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/cluster/DiscoveryEntry.java 2010-06-11 04:11:42 UTC (rev 9306) +++ branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/cluster/DiscoveryEntry.java 2010-06-11 15:33:36 UTC (rev 9307) @@ -13,7 +13,6 @@ package org.hornetq.core.cluster; -import org.hornetq.api.core.Pair; import org.hornetq.api.core.TransportConfiguration; /** @@ -25,19 +24,20 @@ */ public class DiscoveryEntry { - private final Pair<TransportConfiguration, TransportConfiguration> connectorPair; + private final TransportConfiguration connector; private final long lastUpdate; - public DiscoveryEntry(final Pair<TransportConfiguration, TransportConfiguration> connectorPair, final long lastUpdate) + public DiscoveryEntry(final TransportConfiguration connector, final long lastUpdate) { - this.connectorPair = connectorPair; + this.connector = connector; + this.lastUpdate = lastUpdate; } - public Pair<TransportConfiguration, TransportConfiguration> getConnectorPair() + public TransportConfiguration getConnector() { - return connectorPair; + return connector; } public long getLastUpdate() Modified: branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/cluster/impl/DiscoveryGroupImpl.java =================================================================== --- branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/cluster/impl/DiscoveryGroupImpl.java 2010-06-11 04:11:42 UTC (rev 9306) +++ branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/cluster/impl/DiscoveryGroupImpl.java 2010-06-11 15:33:36 UTC (rev 9307) @@ -26,7 +26,6 @@ import org.hornetq.api.core.HornetQBuffer; import org.hornetq.api.core.HornetQBuffers; -import org.hornetq.api.core.Pair; import org.hornetq.api.core.SimpleString; import org.hornetq.api.core.TransportConfiguration; import org.hornetq.api.core.management.NotificationType; @@ -71,7 +70,7 @@ private volatile boolean started; private final String nodeID; - + private final InetAddress localBindAddress; private final InetAddress groupAddress; @@ -94,7 +93,7 @@ this.name = name; this.timeout = timeout; - + this.localBindAddress = localBindAddress; this.groupAddress = groupAddress; @@ -114,7 +113,8 @@ return; } - try { + try + { socket = new MulticastSocket(groupPort); if (localBindAddress != null) @@ -129,10 +129,10 @@ catch (IOException e) { log.error("Failed to create discovery group socket", e); - + return; } - + started = true; thread = new Thread(this, "hornetq-discovery-group-thread-" + name); @@ -144,11 +144,11 @@ if (notificationService != null) { TypedProperties props = new TypedProperties(); - + props.putSimpleStringProperty(new SimpleString("name"), new SimpleString(name)); - + Notification notification = new Notification(nodeID, NotificationType.DISCOVERY_GROUP_STARTED, props); - + notificationService.sendNotification(notification); } } @@ -258,12 +258,11 @@ else { if (!currentUniqueID.equals(uniqueID)) - { - log.warn("There are more than one servers on the network broadcasting the same node id. " + - "You will see this message exactly once (per node) if a node is restarted, in which case it can be safely " + - "ignored. But if it is logged continuously it means you really do have more than one node on the same network " + - "active concurrently with the same node id. This could occur if you have a backup node active at the same time as " + - "its live node."); + { + log.warn("There are more than one servers on the network broadcasting the same node id. " + "You will see this message exactly once (per node) if a node is restarted, in which case it can be safely " + + "ignored. But if it is logged continuously it means you really do have more than one node on the same network " + + "active concurrently with the same node id. This could occur if you have a backup node active at the same time as " + + "its live node."); uniqueIDMap.put(originatingNodeID, uniqueID); } } @@ -308,7 +307,7 @@ String uniqueID = buffer.readString(); checkUniqueID(originatingNodeID, uniqueID); - + if (nodeID.equals(originatingNodeID)) { // Ignore traffic from own node @@ -318,7 +317,7 @@ int size = buffer.readInt(); boolean changed = false; - + synchronized (this) { for (int i = 0; i < size; i++) @@ -326,23 +325,9 @@ TransportConfiguration connector = new TransportConfiguration(); connector.decode(buffer); + + DiscoveryEntry entry = new DiscoveryEntry(connector, System.currentTimeMillis()); - boolean existsBackup = buffer.readBoolean(); - - TransportConfiguration backupConnector = null; - - if (existsBackup) - { - backupConnector = new TransportConfiguration(); - - backupConnector.decode(buffer); - } - - Pair<TransportConfiguration, TransportConfiguration> connectorPair = new Pair<TransportConfiguration, TransportConfiguration>(connector, - backupConnector); - - DiscoveryEntry entry = new DiscoveryEntry(connectorPair, System.currentTimeMillis()); - DiscoveryEntry oldVal = connectors.put(originatingNodeID, entry); if (oldVal == null) Modified: branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/config/BroadcastGroupConfiguration.java =================================================================== --- branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/config/BroadcastGroupConfiguration.java 2010-06-11 04:11:42 UTC (rev 9306) +++ branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/config/BroadcastGroupConfiguration.java 2010-06-11 15:33:36 UTC (rev 9307) @@ -16,7 +16,6 @@ import java.io.Serializable; import java.util.List; -import org.hornetq.api.core.Pair; import org.hornetq.core.logging.Logger; /** @@ -45,7 +44,7 @@ private long broadcastPeriod; - private List<Pair<String, String>> connectorInfos; + private List<String> connectorInfos; public BroadcastGroupConfiguration(final String name, final String localBindAddress, @@ -53,7 +52,7 @@ final String groupAddress, final int groupPort, final long broadcastPeriod, - final List<Pair<String, String>> connectorInfos) + final List<String> connectorInfos) { super(); this.name = name; @@ -95,7 +94,7 @@ return broadcastPeriod; } - public List<Pair<String, String>> getConnectorInfos() + public List<String> getConnectorInfos() { return connectorInfos; } @@ -151,7 +150,7 @@ /** * @param connectorInfos the connectorInfos to set */ - public void setConnectorInfos(final List<Pair<String, String>> connectorInfos) + public void setConnectorInfos(final List<String> connectorInfos) { this.connectorInfos = connectorInfos; } Modified: branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/config/Configuration.java =================================================================== --- branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/config/Configuration.java 2010-06-11 04:11:42 UTC (rev 9306) +++ branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/config/Configuration.java 2010-06-11 15:33:36 UTC (rev 9307) @@ -258,16 +258,15 @@ void setConnectorConfigurations(Map<String, TransportConfiguration> infos); /** - * Returns the name of the connector used to connect to the backup. - * <br> - * If this server has no backup or is itself a backup, the value is {@code null}. + * Returns the name of the connector used to connect to the live node - only used when using shared nothing (shared store = false). + * <br> */ - String getBackupConnectorName(); + String getLiveConnectorName(); /** - * Sets the name of the connector used to connect to the backup. + * Sets the name of the connector used to connect to the live node - only used when using shared nothing (shared store = false). */ - void setBackupConnectorName(String name); + void setLiveConnectorName(String name); /** * Returns the broadcast groups configured for this server. Modified: branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/config/impl/ConfigurationImpl.java =================================================================== --- branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/config/impl/ConfigurationImpl.java 2010-06-11 04:11:42 UTC (rev 9306) +++ branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/config/impl/ConfigurationImpl.java 2010-06-11 15:33:36 UTC (rev 9307) @@ -225,7 +225,7 @@ protected Set<TransportConfiguration> acceptorConfigs = new HashSet<TransportConfiguration>(); - protected String backupConnectorName; + protected String liveConnectorName; protected List<BridgeConfiguration> bridgeConfigurations = new ArrayList<BridgeConfiguration>(); @@ -477,14 +477,14 @@ connectorConfigs = infos; } - public String getBackupConnectorName() + public String getLiveConnectorName() { - return backupConnectorName; + return liveConnectorName; } - public void setBackupConnectorName(final String backupConnectorName) + public void setLiveConnectorName(final String liveConnectorName) { - this.backupConnectorName = backupConnectorName; + this.liveConnectorName = liveConnectorName; } public GroupingHandlerConfiguration getGroupingHandlerConfiguration() @@ -1015,14 +1015,14 @@ { return false; } - if (backupConnectorName == null) + if (liveConnectorName == null) { - if (other.backupConnectorName != null) + if (other.liveConnectorName != null) { return false; } } - else if (!backupConnectorName.equals(other.backupConnectorName)) + else if (!liveConnectorName.equals(other.liveConnectorName)) { return false; } Modified: branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/config/impl/FileConfiguration.java =================================================================== --- branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/config/impl/FileConfiguration.java 2010-06-11 04:11:42 UTC (rev 9306) +++ branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/config/impl/FileConfiguration.java 2010-06-11 15:33:36 UTC (rev 9307) @@ -47,7 +47,7 @@ // Attributes ---------------------------------------------------------------------- - private String configurationUrl = FileConfiguration.DEFAULT_CONFIGURATION_URL; + private String configurationUrl = DEFAULT_CONFIGURATION_URL; private boolean started; Modified: branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/deployers/impl/FileConfigurationParser.java =================================================================== --- branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/deployers/impl/FileConfigurationParser.java 2010-06-11 04:11:42 UTC (rev 9306) +++ branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/deployers/impl/FileConfigurationParser.java 2010-06-11 15:33:36 UTC (rev 9307) @@ -31,15 +31,14 @@ import org.hornetq.core.config.BroadcastGroupConfiguration; import org.hornetq.core.config.ClusterConnectionConfiguration; import org.hornetq.core.config.Configuration; +import org.hornetq.core.config.CoreQueueConfiguration; import org.hornetq.core.config.DiscoveryGroupConfiguration; import org.hornetq.core.config.DivertConfiguration; -import org.hornetq.core.config.CoreQueueConfiguration; import org.hornetq.core.config.impl.ConfigurationImpl; import org.hornetq.core.config.impl.FileConfiguration; import org.hornetq.core.config.impl.Validators; import org.hornetq.core.journal.impl.AIOSequentialFileFactory; import org.hornetq.core.logging.Logger; -import org.hornetq.core.persistence.impl.journal.JournalStorageManager; import org.hornetq.core.security.Role; import org.hornetq.core.server.JournalType; import org.hornetq.core.server.group.impl.GroupingHandlerConfiguration; @@ -278,15 +277,15 @@ config.setInterceptorClassNames(interceptorList); - NodeList backups = e.getElementsByTagName("backup-connector-ref"); + NodeList lives = e.getElementsByTagName("live-connector-ref"); - // There should be only one - this will be enforced by the DTD + // There should be at most one - this will be enforced by the DTD - if (backups.getLength() > 0) + if (lives.getLength() > 0) { - Node backupNode = backups.item(0); + Node liveNode = lives.item(0); - config.setBackupConnectorName(backupNode.getAttributes().getNamedItem("connector-name").getNodeValue()); + config.setLiveConnectorName(liveNode.getAttributes().getNamedItem("connector-name").getNodeValue()); } NodeList connectorNodes = e.getElementsByTagName("connector"); @@ -872,7 +871,7 @@ NodeList children = e.getChildNodes(); - List<Pair<String, String>> connectorNames = new ArrayList<Pair<String, String>>(); + List<String> connectorNames = new ArrayList<String>(); for (int j = 0; j < children.getLength(); j++) { @@ -882,18 +881,7 @@ { String connectorName = child.getAttributes().getNamedItem("connector-name").getNodeValue(); - Node backupConnectorNode = child.getAttributes().getNamedItem("backup-connector-name"); - - String backupConnectorName = null; - - if (backupConnectorNode != null) - { - backupConnectorName = backupConnectorNode.getNodeValue(); - } - - Pair<String, String> connectorInfo = new Pair<String, String>(connectorName, backupConnectorName); - - connectorNames.add(connectorInfo); + connectorNames.add(connectorName); } } Modified: branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/management/impl/BroadcastGroupControlImpl.java =================================================================== --- branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/management/impl/BroadcastGroupControlImpl.java 2010-06-11 04:11:42 UTC (rev 9306) +++ branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/management/impl/BroadcastGroupControlImpl.java 2010-06-11 15:33:36 UTC (rev 9307) @@ -15,14 +15,11 @@ import javax.management.MBeanOperationInfo; -import org.hornetq.api.core.Pair; -import org.hornetq.api.core.management.AddressControl; import org.hornetq.api.core.management.BroadcastGroupControl; import org.hornetq.core.config.BroadcastGroupConfiguration; import org.hornetq.core.persistence.StorageManager; import org.hornetq.core.server.cluster.BroadcastGroup; import org.hornetq.utils.json.JSONArray; -import org.hornetq.utils.json.JSONObject; /** * A BroadcastGroupControl @@ -91,14 +88,9 @@ Object[] ret = new Object[configuration.getConnectorInfos().size()]; int i = 0; - for (Pair<String, String> pair : configuration.getConnectorInfos()) + for (String connector : configuration.getConnectorInfos()) { - String[] opair = new String[2]; - - opair[0] = pair.a; - opair[1] = pair.b != null ? pair.b : null; - - ret[i++] = opair; + ret[i++] = connector; } return ret; @@ -116,12 +108,9 @@ { JSONArray array = new JSONArray(); - for (Pair<String, String> pair : configuration.getConnectorInfos()) + for (String connector : configuration.getConnectorInfos()) { - JSONObject p = new JSONObject(); - p.put("a", pair.a); - p.put("b", pair.b); - array.put(p); + array.put(connector); } return array.toString(); } Modified: branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/management/impl/HornetQServerControlImpl.java =================================================================== --- branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/management/impl/HornetQServerControlImpl.java 2010-06-11 04:11:42 UTC (rev 9306) +++ branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/management/impl/HornetQServerControlImpl.java 2010-06-11 15:33:36 UTC (rev 9307) @@ -175,14 +175,14 @@ } } - public String getBackupConnectorName() + public String getLiveConnectorName() { checkStarted(); clearIO(); try { - return configuration.getBackupConnectorName(); + return configuration.getLiveConnectorName(); } finally { Modified: branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/persistence/StorageManager.java =================================================================== --- branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/persistence/StorageManager.java 2010-06-11 04:11:42 UTC (rev 9306) +++ branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/persistence/StorageManager.java 2010-06-11 15:33:36 UTC (rev 9307) @@ -13,7 +13,6 @@ package org.hornetq.core.persistence; -import java.util.ArrayList; import java.util.List; import java.util.Map; import java.util.concurrent.Executor; @@ -82,11 +81,7 @@ void waitOnOperations() throws Exception; void clearContext(); - - UUID getPersistentID(); - - void setPersistentID(UUID id) throws Exception; - + long generateUniqueID(); long getCurrentUniqueID(); Modified: branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java =================================================================== --- branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java 2010-06-11 04:11:42 UTC (rev 9306) +++ branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/persistence/impl/journal/JournalStorageManager.java 2010-06-11 15:33:36 UTC (rev 9307) @@ -100,8 +100,6 @@ public static final byte QUEUE_BINDING_RECORD = 21; - public static final byte PERSISTENT_ID_RECORD = 23; - public static final byte ID_COUNTER_RECORD = 24; public static final byte ADDRESS_SETTING_RECORD = 25; @@ -133,8 +131,6 @@ public static final byte HEURISTIC_COMPLETION = 38; - private UUID persistentID; - private final BatchingIDGenerator idGenerator; private final ReplicationManager replicator; @@ -261,14 +257,7 @@ throw new IllegalArgumentException("Unsupported journal type " + config.getJournalType()); } - if (config.isBackup()) - { - idGenerator = null; - } - else - { - idGenerator = new BatchingIDGenerator(0, JournalStorageManager.CHECKPOINT_BATCH_SIZE, bindingsJournal); - } + idGenerator = new BatchingIDGenerator(0, JournalStorageManager.CHECKPOINT_BATCH_SIZE, bindingsJournal); Journal localMessage = new JournalImpl(config.getJournalFileSize(), config.getJournalMinFiles(), @@ -393,26 +382,6 @@ getContext().executeOnCompletion(run); } - public UUID getPersistentID() - { - return persistentID; - } - - public void setPersistentID(final UUID id) throws Exception - { - long recordID = generateUniqueID(); - - if (id != null) - { - bindingsJournal.appendAddRecord(recordID, - JournalStorageManager.PERSISTENT_ID_RECORD, - new PersistentIDEncoding(id), - true); - } - - persistentID = id; - } - public long generateUniqueID() { long id = idGenerator.generateID(); @@ -1079,14 +1048,6 @@ queueBindingInfos.add(bindingEncoding); } - else if (rec == JournalStorageManager.PERSISTENT_ID_RECORD) - { - PersistentIDEncoding encoding = new PersistentIDEncoding(); - - encoding.decode(buffer); - - persistentID = encoding.uuid; - } else if (rec == JournalStorageManager.ID_COUNTER_RECORD) { idGenerator.loadState(record.id, buffer); @@ -1154,17 +1115,12 @@ } // Must call close to make sure last id is persisted - if (idGenerator != null) - { - idGenerator.close(); - } + idGenerator.close(); bindingsJournal.stop(); messageJournal.stop(); - persistentID = null; - started = false; } Modified: branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/persistence/impl/nullpm/NullStorageManager.java =================================================================== --- branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/persistence/impl/nullpm/NullStorageManager.java 2010-06-11 04:11:42 UTC (rev 9306) +++ branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/persistence/impl/nullpm/NullStorageManager.java 2010-06-11 15:33:36 UTC (rev 9307) @@ -13,7 +13,6 @@ package org.hornetq.core.persistence.impl.nullpm; -import java.util.ArrayList; import java.util.Collections; import java.util.List; import java.util.Map; @@ -46,7 +45,6 @@ import org.hornetq.core.server.ServerMessage; import org.hornetq.core.server.group.impl.GroupBinding; import org.hornetq.core.transaction.ResourceManager; -import org.hornetq.utils.UUID; /** * @@ -60,20 +58,8 @@ { private final AtomicLong idSequence = new AtomicLong(0); - private UUID id; - private volatile boolean started; - public UUID getPersistentID() - { - return id; - } - - public void setPersistentID(final UUID id) - { - this.id = id; - } - public void sync() { // NO OP @@ -247,8 +233,6 @@ throw new IllegalStateException("Not started"); } - id = null; - idSequence.set(0); started = false; Modified: branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/protocol/core/impl/PacketDecoder.java =================================================================== --- branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/protocol/core/impl/PacketDecoder.java 2010-06-11 04:11:42 UTC (rev 9306) +++ branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/protocol/core/impl/PacketDecoder.java 2010-06-11 15:33:36 UTC (rev 9307) @@ -13,6 +13,7 @@ package org.hornetq.core.protocol.core.impl; +import static org.hornetq.core.protocol.core.impl.PacketImpl.CLUSTER_TOPOLOGY; import static org.hornetq.core.protocol.core.impl.PacketImpl.CREATESESSION; import static org.hornetq.core.protocol.core.impl.PacketImpl.CREATESESSION_RESP; import static org.hornetq.core.protocol.core.impl.PacketImpl.CREATE_QUEUE; @@ -82,6 +83,7 @@ import org.hornetq.api.core.HornetQBuffer; import org.hornetq.core.logging.Logger; import org.hornetq.core.protocol.core.Packet; +import org.hornetq.core.protocol.core.impl.wireformat.ClusterTopologyMessage; import org.hornetq.core.protocol.core.impl.wireformat.CreateQueueMessage; import org.hornetq.core.protocol.core.impl.wireformat.CreateReplicationSessionMessage; import org.hornetq.core.protocol.core.impl.wireformat.CreateSessionMessage; @@ -486,6 +488,11 @@ packet = new SessionForceConsumerDelivery(); break; } + case CLUSTER_TOPOLOGY: + { + packet = new ClusterTopologyMessage(); + break; + } default: { throw new IllegalArgumentException("Invalid type: " + packetType); Modified: branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/protocol/core/impl/PacketImpl.java =================================================================== --- branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/protocol/core/impl/PacketImpl.java 2010-06-11 04:11:42 UTC (rev 9306) +++ branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/protocol/core/impl/PacketImpl.java 2010-06-11 15:33:36 UTC (rev 9307) @@ -181,6 +181,10 @@ public static final byte REPLICATION_COMPARE_DATA = 102; public static final byte REPLICATION_SYNC = 103; + + // HA + + public static final byte CLUSTER_TOPOLOGY = 110; // Static -------------------------------------------------------- Modified: branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/server/cluster/BroadcastGroup.java =================================================================== --- branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/server/cluster/BroadcastGroup.java 2010-06-11 04:11:42 UTC (rev 9306) +++ branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/server/cluster/BroadcastGroup.java 2010-06-11 15:33:36 UTC (rev 9307) @@ -13,7 +13,6 @@ package org.hornetq.core.server.cluster; -import org.hornetq.api.core.Pair; import org.hornetq.api.core.TransportConfiguration; import org.hornetq.core.server.HornetQComponent; import org.hornetq.core.server.management.NotificationService; @@ -34,9 +33,9 @@ String getName(); - void addConnectorPair(Pair<TransportConfiguration, TransportConfiguration> connectorPair); + void addConnector(TransportConfiguration tcConfig); - void removeConnectorPair(Pair<TransportConfiguration, TransportConfiguration> connectorPair); + void removeConnector(TransportConfiguration tcConfig); int size(); Modified: branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/server/cluster/ClusterManager.java =================================================================== --- branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/server/cluster/ClusterManager.java 2010-06-11 04:11:42 UTC (rev 9306) +++ branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/server/cluster/ClusterManager.java 2010-06-11 15:33:36 UTC (rev 9307) @@ -13,10 +13,13 @@ package org.hornetq.core.server.cluster; +import java.util.List; import java.util.Map; import java.util.Set; +import org.hornetq.api.core.Pair; import org.hornetq.api.core.SimpleString; +import org.hornetq.api.core.TransportConfiguration; import org.hornetq.core.server.HornetQComponent; /** @@ -38,4 +41,14 @@ Set<BroadcastGroup> getBroadcastGroups(); void activate(); + + void startAnnouncement(); + + void stopAnnouncement(); + + List<Pair<TransportConfiguration, TransportConfiguration>> getClusterTopology(); + + void registerTopologyListener(ClusterTopologyListener listener); + + void unregisterTopologyListener(ClusterTopologyListener listener); } Modified: branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/server/cluster/impl/BroadcastGroupImpl.java =================================================================== --- branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/server/cluster/impl/BroadcastGroupImpl.java 2010-06-11 04:11:42 UTC (rev 9306) +++ branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/server/cluster/impl/BroadcastGroupImpl.java 2010-06-11 15:33:36 UTC (rev 9307) @@ -22,7 +22,6 @@ import org.hornetq.api.core.HornetQBuffer; import org.hornetq.api.core.HornetQBuffers; -import org.hornetq.api.core.Pair; import org.hornetq.api.core.SimpleString; import org.hornetq.api.core.TransportConfiguration; import org.hornetq.api.core.management.NotificationType; @@ -59,7 +58,7 @@ private DatagramSocket socket; - private final List<Pair<TransportConfiguration, TransportConfiguration>> connectorPairs = new ArrayList<Pair<TransportConfiguration, TransportConfiguration>>(); + private final List<TransportConfiguration> connectors = new ArrayList<TransportConfiguration>(); private boolean started; @@ -181,19 +180,19 @@ return name; } - public synchronized void addConnectorPair(final Pair<TransportConfiguration, TransportConfiguration> connectorPair) + public synchronized void addConnector(final TransportConfiguration tcConfig) { - connectorPairs.add(connectorPair); + connectors.add(tcConfig); } - public synchronized void removeConnectorPair(final Pair<TransportConfiguration, TransportConfiguration> connectorPair) + public synchronized void removeConnector(final TransportConfiguration tcConfig) { - connectorPairs.remove(connectorPair); + connectors.remove(tcConfig); } public synchronized int size() { - return connectorPairs.size(); + return connectors.size(); } public synchronized void activate() @@ -214,22 +213,11 @@ buff.writeString(uniqueID); - buff.writeInt(connectorPairs.size()); + buff.writeInt(connectors.size()); - for (Pair<TransportConfiguration, TransportConfiguration> connectorPair : connectorPairs) + for (TransportConfiguration tcConfig : connectors) { - connectorPair.a.encode(buff); - - if (connectorPair.b != null) - { - buff.writeBoolean(true); - - connectorPair.b.encode(buff); - } - else - { - buff.writeBoolean(false); - } + tcConfig.encode(buff); } byte[] data = buff.toByteBuffer().array(); Modified: branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/server/cluster/impl/ClusterManagerImpl.java =================================================================== --- branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/server/cluster/impl/ClusterManagerImpl.java 2010-06-11 04:11:42 UTC (rev 9306) +++ branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/server/cluster/impl/ClusterManagerImpl.java 2010-06-11 15:33:36 UTC (rev 9307) @@ -44,8 +44,11 @@ import org.hornetq.core.server.cluster.BroadcastGroup; import org.hornetq.core.server.cluster.ClusterConnection; import org.hornetq.core.server.cluster.ClusterManager; +import org.hornetq.core.server.cluster.ClusterTopologyListener; import org.hornetq.core.server.cluster.Transformer; import org.hornetq.core.server.management.ManagementService; +import org.hornetq.utils.ConcurrentHashSet; +import org.hornetq.utils.ExecutorFactory; import org.hornetq.utils.UUID; /** @@ -67,9 +70,9 @@ private final Map<String, Bridge> bridges = new HashMap<String, Bridge>(); - private final Map<String, ClusterConnection> clusters = new HashMap<String, ClusterConnection>(); + private final Map<String, ClusterConnection> clusterConnections = new HashMap<String, ClusterConnection>(); - private final org.hornetq.utils.ExecutorFactory executorFactory; + private final ExecutorFactory executorFactory; private final HornetQServer server; @@ -89,7 +92,7 @@ private final boolean clustered; - public ClusterManagerImpl(final org.hornetq.utils.ExecutorFactory executorFactory, + public ClusterManagerImpl(final ExecutorFactory executorFactory, final HornetQServer server, final PostOffice postOffice, final ScheduledExecutorService scheduledExecutor, @@ -177,7 +180,7 @@ managementService.unregisterDiscoveryGroup(group.getName()); } - for (ClusterConnection clusterConnection : clusters.values()) + for (ClusterConnection clusterConnection : clusterConnections.values()) { clusterConnection.stop(); managementService.unregisterCluster(clusterConnection.getName().toString()); @@ -211,7 +214,7 @@ public Set<ClusterConnection> getClusterConnections() { - return new HashSet<ClusterConnection>(clusters.values()); + return new HashSet<ClusterConnection>(clusterConnections.values()); } public Set<BroadcastGroup> getBroadcastGroups() @@ -221,7 +224,7 @@ public ClusterConnection getClusterConnection(final SimpleString name) { - return clusters.get(name.toString()); + return clusterConnections.get(name.toString()); } public synchronized void activate() @@ -236,13 +239,42 @@ bridge.activate(); } - for (ClusterConnection cc : clusters.values()) + for (ClusterConnection cc : clusterConnections.values()) { cc.activate(); } backup = false; } + + public void startAnnouncement() + { + + } + + public void stopAnnouncement() + { + + } + + private Set<ClusterTopologyListener> listeners = new ConcurrentHashSet<ClusterTopologyListener>(); + + private List<Pair<TransportConfiguration, TransportConfiguration>> topology; + + public List<Pair<TransportConfiguration, TransportConfiguration>> getClusterTopology() + { + return topology; + } + + public void registerTopologyListener(final ClusterTopologyListener listener) + { + listeners.add(listener); + } + + public void unregisterTopologyListener(final ClusterTopologyListener listener) + { + listeners.remove(listener); + } private synchronized void deployBroadcastGroup(final BroadcastGroupConfiguration config) throws Exception { @@ -270,32 +302,18 @@ config.getGroupPort(), !backup); - for (Pair<String, String> connectorInfo : config.getConnectorInfos()) + for (String connectorInfo : config.getConnectorInfos()) { - TransportConfiguration connector = configuration.getConnectorConfigurations().get(connectorInfo.a); + TransportConfiguration connector = configuration.getConnectorConfigurations().get(connectorInfo); if (connector == null) { - logWarnNoConnector(config.getName(), connectorInfo.a); + logWarnNoConnector(config.getName(), connectorInfo); return; } - TransportConfiguration backupConnector = null; - - if (connectorInfo.b != null) - { - backupConnector = configuration.getConnectorConfigurations().get(connectorInfo.b); - - if (connector == null) - { - logWarnNoConnector(config.getName(), connectorInfo.b); - - return; - } - } - - group.addConnectorPair(new Pair<TransportConfiguration, TransportConfiguration>(connector, backupConnector)); + group.addConnector(connector); } ScheduledFuture<?> future = scheduledExecutor.scheduleWithFixedDelay(group, @@ -607,7 +625,7 @@ managementService.registerCluster(clusterConnection, config); - clusters.put(config.getName(), clusterConnection); + clusterConnections.put(config.getName(), clusterConnection); clusterConnection.start(); } Modified: branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java =================================================================== --- branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2010-06-11 04:11:42 UTC (rev 9306) +++ branches/2_2_0_HA_Improvements/src/main/org/hornetq/core/server/impl/HornetQServerImpl.java 2010-06-11 15:33:36 UTC (rev 9307) @@ -13,6 +13,13 @@ package org.hornetq.core.server.impl; +import java.io.BufferedInputStream; +import java.io.BufferedOutputStream; +import java.io.File; +import java.io.FileInputStream; +import java.io.FileOutputStream; +import java.io.InputStream; +import java.io.OutputStream; import java.lang.management.ManagementFactory; import java.security.AccessController; import java.security.PrivilegedAction; @@ -40,8 +47,6 @@ import org.hornetq.api.core.client.ClientSessionFactory; import org.hornetq.api.core.client.HornetQClient; import org.hornetq.core.client.impl.ClientSessionFactoryImpl; -import org.hornetq.core.client.impl.FailoverManager; -import org.hornetq.core.client.impl.FailoverManagerImpl; import org.hornetq.core.config.Configuration; import org.hornetq.core.config.CoreQueueConfiguration; import org.hornetq.core.config.DivertConfiguration; @@ -80,8 +85,6 @@ import org.hornetq.core.remoting.server.impl.RemotingServiceImpl; import org.hornetq.core.replication.ReplicationEndpoint; import org.hornetq.core.replication.ReplicationManager; -import org.hornetq.core.replication.impl.ReplicationEndpointImpl; -import org.hornetq.core.replication.impl.ReplicationManagerImpl; import org.hornetq.core.security.CheckType; import org.hornetq.core.security.Role; import org.hornetq.core.security.SecurityStore; @@ -94,8 +97,10 @@ import org.hornetq.core.server.QueueFactory; import org.hornetq.core.server.ServerSession; import org.hornetq.core.server.cluster.ClusterManager; +import org.hornetq.core.server.cluster.LockFile; import org.hornetq.core.server.cluster.Transformer; import org.hornetq.core.server.cluster.impl.ClusterManagerImpl; +import org.hornetq.core.server.cluster.impl.LockFileImpl; import org.hornetq.core.server.group.GroupingHandler; import org.hornetq.core.server.group.impl.GroupBinding; import org.hornetq.core.server.group.impl.GroupingHandlerConfiguration; @@ -274,6 +279,329 @@ // lifecycle methods // ---------------------------------------------------------------- + private interface Activation extends Runnable + { + void close() throws Exception; + } + + /* + * Can be overridden for tests + */ + protected LockFile createLockFile(final String fileName, final String directory) + { + return new LockFileImpl(fileName, directory); + } + + private class SharedStoreLiveActivation implements Activation + { + LockFile liveLock; + + public void run() + { + try + { + log.info("Waiting to obtain live lock"); + + File journalDir = new File(configuration.getJournalDirectory()); + + if (!journalDir.exists()) + { + if (configuration.isCreateJournalDir()) + { + journalDir.mkdirs(); + } + else + { + throw new IllegalArgumentException("Directory " + journalDir + + " does not exist and will not create it"); + } + } + + liveLock = createLockFile("live.lock", configuration.getJournalDirectory()); + + liveLock.lock(); + + log.info("Obtained live lock"); + + // We now load the node id file, creating it, if it doesn't exist yet + File nodeIDFile = new File(configuration.getJournalDirectory(), "node.id"); + + if (!nodeIDFile.exists()) + { + // We use another file lock to prevent a backup reading it before it is complete + + LockFile nodeIDLockFile = createLockFile("nodeid.lock", configuration.getJournalDirectory()); + + nodeIDLockFile.lock(); + + OutputStream os = null; + + try + { + os = new BufferedOutputStream(new FileOutputStream(nodeIDFile)); + + uuid = UUIDGenerator.getInstance().generateUUID(); + + nodeID = new SimpleString(uuid.toString()); + + os.write(uuid.asBytes()); + + log.info("Wrote node id, it is " + nodeID); + } + finally + { + if (os != null) + { + os.close(); + } + } + + nodeIDLockFile.unlock(); + } + else + { + // Read it + + readNodeID(nodeIDFile); + } + + initialisePart1(); + + initialisePart2(); + + log.info("Server is now live"); + } + catch (Exception e) + { + log.error("Failure in initialisation", e); + } + } + + public void close() throws Exception + { + if (liveLock != null) + { + // We need to delete the file too, otherwise the backup will failover when we shutdown or if the backup is + // started before the live + + File liveFile = new File(configuration.getJournalDirectory(), "live.lock"); + + liveFile.delete(); + + liveLock.unlock(); + + } + } + } + + private void readNodeID(final File nodeIDFile) throws Exception + { + // Read it + InputStream is = null; + + try + { + is = new BufferedInputStream(new FileInputStream(nodeIDFile)); + + byte[] bytes = new byte[16]; + + int read = 0; + + while (read < 16) + { + int r = is.read(bytes, read, 16 - read); + + if (r <= 0) + { + throw new IllegalStateException("Cannot read node id file, perhaps it is corrupt?"); + } + + read += r; + } + + uuid = new UUID(UUID.TYPE_TIME_BASED, bytes); + + nodeID = new SimpleString(uuid.toString()); + + log.info("Read node id, it is " + nodeID); + } + finally + { + if (is != null) + { + is.close(); + } + } + } + + private class SharedStoreBackupActivation implements Activation + { + LockFile backupLock; + + LockFile liveLock; + + public void run() + { + try + { + backupLock = createLockFile("backup.lock", configuration.getJournalDirectory()); + + log.info("Waiting to become backup node"); + + backupLock.lock(); + + log.info("** got backup lock"); + + // We load the node id from the file in the journal dir - if the backup is started before live and live has + // never been started before it may not exist yet, so + // we wait for it + + File nodeIDFile = new File(configuration.getJournalDirectory(), "node.id"); + + while (true) + { + // We also need to create another lock file for the node.id file since we don't want to see any partially + // written + // node id if the live node is still creating it. + // Also renaming is not atomic necessarily so we can't use a write and rename strategy safely + + LockFile nodeIDLockFile = createLockFile("nodeid.lock", configuration.getJournalDirectory()); + + nodeIDLockFile.lock(); + + if (!nodeIDFile.exists()) + { + nodeIDLockFile.unlock(); + + Thread.sleep(2000); + + continue; + } + + nodeIDLockFile.unlock(); + + break; + } + + readNodeID(nodeIDFile); + + log.info("Read node id " + nodeID); + + initialisePart1(); + + // TODO - now send announcement message to cluster + + // We now look for the live.lock file - if it doesn't exist it means the live isn't started yet, so we wait + // for that + + clusterManager.startAnnouncement(); + + while (true) + { + File liveLockFile = new File(configuration.getJournalDirectory(), "live.lock"); + + while (!liveLockFile.exists()) + { + log.info("Waiting for server live lock file. Live server is not started"); + + Thread.sleep(2000); + } + + liveLock = createLockFile("live.lock", configuration.getJournalDirectory()); + + log.info("Live server is up - waiting for failover"); + + liveLock.lock(); + + // We need to test if the file exists again, since the live might have shutdown + if (!liveLockFile.exists()) + { + liveLock.unlock(); + + continue; + } + + log.info("Obtained live lock"); + + break; + } + + configuration.setBackup(false); + + initialisePart2(); + + log.info("Server is now live"); + + backupLock.unlock(); + } + catch (InterruptedException e) + { + // This can occur when closing if the thread is blocked - it's ok + } + catch (Exception e) + { + log.error("Failure in initialisation", e); + } + } + + public void close() throws Exception + { + long timeout = 30000; + + long start = System.currentTimeMillis(); + + while (backupActivationThread.isAlive() && System.currentTimeMillis() - start < timeout) + { + backupActivationThread.interrupt(); + + Thread.sleep(1000); + } + + if (System.currentTimeMillis() - start >= timeout) + { + log.warn("Timed out waiting for backup activation to exit"); + } + + if (liveLock != null) + { + liveLock.unlock(); + } + + if (backupLock != null) + { + backupLock.unlock(); + } + + } + } + + private class SharedNothingBackupActivation implements Activation + { + public void run() + { + try + { + // TODO + + // Try-Connect to live server using live-connector-ref + + // sit in loop and try and connect, if server is not live then it will return NOT_LIVE + } + catch (Exception e) + { + log.error("Failure in initialisation", e); + } + } + + public void close() throws Exception + { + } + } + + private Thread backupActivationThread; + + private Activation activation; + public synchronized void start() throws Exception { initialiseLogging(); @@ -293,30 +621,66 @@ test.run(); } - initialisePart1(); + if (!configuration.isBackup()) + { + if (configuration.isSharedStore()) + { + activation = new SharedStoreLiveActivation(); + // This should block until the lock is got + + activation.run(); + } + } + + started = true; + + HornetQServerImpl.log.info("HornetQ Server version " + getVersion().getFullVersion() + " started"); + if (configuration.isBackup()) { - if (!configuration.isSharedStore()) + if (configuration.isSharedStore()) { - replicationEndpoint = new ReplicationEndpointImpl(this); - replicationEndpoint.start(); + activation = new SharedStoreBackupActivation(); } - // We defer actually initialisation until the live node has contacted the backup - HornetQServerImpl.log.info("Backup server initialised"); + else + { + // Replicated + + activation = new SharedNothingBackupActivation(); + } + + backupActivationThread = new Thread(activation); + + backupActivationThread.start(); } - else - { - initialisePart2(); - } - // We start the remoting service here - if the server is a backup remoting service needs to be started - // so it can be initialised by the live node - remotingService.start(); + // initialisePart1(); + // + // if (configuration.isBackup()) + // { + // if (!configuration.isSharedStore()) + // { + // replicationEndpoint = new ReplicationEndpointImpl(this); + // replicationEndpoint.start(); + // } + // else + // { + // backupLock = new FailoverLockFileImpl("backup.lock", configuration.getJournalDirectory()); + // liveLock = new FailoverLockFileImpl("live.lock", configuration.getJournalDirectory()); + // } + // + // // We defer actually initialisation until the live node has contacted the backup + // //HornetQServerImpl.log.info("Backup server initialised"); + // } + // else + // { + // initialisePart2(); + // } + // + // + // remotingService.start(); - started = true; - - HornetQServerImpl.log.info("HornetQ Server version " + getVersion().getFullVersion() + " started"); } @Override @@ -334,6 +698,10 @@ public void stop() throws Exception { + System.out.println("*** stop called on server"); + + System.out.flush(); + synchronized (this) { if (!started) @@ -351,12 +719,6 @@ managementService.removeNotificationListener(groupingHandler); groupingHandler = null; } - // // Need to flush all sessions to make sure all confirmations get sent back to client - // - // for (ServerSession session : sessions.values()) - // { - // session.getChannel().flushConfirmations(); - // } } // we stop the remoting service outside a lock @@ -459,9 +821,19 @@ started = false; initialised = false; - uuid = null; + // uuid = null; nodeID = null; + if (activation != null) + { + activation.close(); + } + + if (backupActivationThread != null) + { + backupActivationThread.join(); + } + HornetQServerImpl.log.info("HornetQ Server version " + getVersion().getFullVersion() + " stopped"); Logger.reset(); @@ -584,8 +956,8 @@ managementService, this, configuration.getManagementAddress(), - defaultAddress == null ? null : - new SimpleString(defaultAddress), + defaultAddress == null ? null + : new SimpleString(defaultAddress), callback); sessions.put(name, session); @@ -685,6 +1057,8 @@ final boolean durable, final boolean temporary) throws Exception { + log.info("trying to deploy queue " + queueName); + return createQueue(address, queueName, filterString, durable, temporary, true); } @@ -760,7 +1134,7 @@ { return replicationEndpoint; } - + public ReplicationManager getReplicationManager() { return replicationManager; @@ -793,7 +1167,7 @@ 0, 1.0d, 0, - 1, + 1, false, threadPool, scheduledPool, @@ -827,34 +1201,34 @@ // Private // -------------------------------------------------------------------------------------- - private boolean startReplication() throws Exception - { - String backupConnectorName = configuration.getBackupConnectorName(); + // private boolean startReplication() throws Exception + // { + // String backupConnectorName = configuration.getBackupConnectorName(); + // + // if (!configuration.isSharedStore() && backupConnectorName != null) + // { + // TransportConfiguration backupConnector = configuration.getConnectorConfigurations().get(backupConnectorName); + // + // if (backupConnector == null) + // { + // HornetQServerImpl.log.warn("connector with name '" + backupConnectorName + + // "' is not defined in the configuration."); + // } + // else + // { + // + // replicationFailoverManager = createBackupConnectionFailoverManager(backupConnector, + // threadPool, + // scheduledPool); + // + // replicationManager = new ReplicationManagerImpl(replicationFailoverManager, executorFactory); + // replicationManager.start(); + // } + // } + // + // return true; + // } - if (!configuration.isSharedStore() && backupConnectorName != null) - { - TransportConfiguration backupConnector = configuration.getConnectorConfigurations().get(backupConnectorName); - - if (backupConnector == null) - { - HornetQServerImpl.log.warn("connector with name '" + backupConnectorName + - "' is not defined in the configuration."); - } - else - { - - replicationFailoverManager = createBackupConnectionFailoverManager(backupConnector, - threadPool, - scheduledPool); - - replicationManager = new ReplicationManagerImpl(replicationFailoverManager, executorFactory); - replicationManager.start(); - } - } - - return true; - } - private void callActivateCallbacks() { for (ActivateCallback callback : activateCallbacks) @@ -863,7 +1237,6 @@ } } - private void callPreActiveCallbacks() { for (ActivateCallback callback : activateCallbacks) @@ -902,11 +1275,31 @@ return true; } + private class FileActivateRunner implements Runnable + { + public void run() + { + + } + } + + private void initialiseLogging() + { + LogDelegateFactory logDelegateFactory = (LogDelegateFactory)instantiateInstance(configuration.getLogDelegateFactoryClassName()); + + Logger.setDelegateFactory(logDelegateFactory); + } + + /* + * Start everything apart from RemotingService and loading the data + */ private void initialisePart1() throws Exception { // Create the pools - we have two pools - one for non scheduled - and another for scheduled - ThreadFactory tFactory = new HornetQThreadFactory("HornetQ-server-threads" + System.identityHashCode(this), false, getThisClassLoader()); + ThreadFactory tFactory = new HornetQThreadFactory("HornetQ-server-threads" + System.identityHashCode(this), + false, + getThisClassLoader()); if (configuration.getThreadPoolMaxSize() == -1) { @@ -920,7 +1313,9 @@ executorFactory = new OrderedExecutorFactory(threadPool); scheduledPool = new ScheduledThreadPoolExecutor(configuration.getScheduledThreadPoolMaxSize(), - new HornetQThreadFactory("HornetQ-scheduled-threads", false, getThisClassLoader())); + new HornetQThreadFactory("HornetQ-scheduled-threads", + false, + getThisClassLoader())); managementService = new ManagementServiceImpl(mbeanServer, configuration); @@ -933,27 +1328,17 @@ memoryManager.start(); } - } - private void initialiseLogging() - { - LogDelegateFactory logDelegateFactory = (LogDelegateFactory)instantiateInstance(configuration.getLogDelegateFactoryClassName()); - - Logger.setDelegateFactory(logDelegateFactory); - } - - private void initialisePart2() throws Exception - { // Create the hard-wired components if (configuration.isFileDeploymentEnabled()) { deploymentManager = new FileDeploymentManager(configuration.getFileDeployerScanPeriod()); } - + callPreActiveCallbacks(); - startReplication(); + // startReplication(); storageManager = createStorageManager(); @@ -1049,14 +1434,52 @@ { deploySecurityFromConfiguration(); } - + deployGroupingHandlerConfiguration(configuration.getGroupingHandlerConfiguration()); + // This can't be created until node id is set + clusterManager = new ClusterManagerImpl(executorFactory, + this, + postOffice, + scheduledPool, + managementService, + configuration, + uuid, + configuration.isBackup(), + configuration.isClustered()); + + } + + /* + * Load the data, and start remoting service so clients can connect + */ + private void initialisePart2() throws Exception + { // Load the journal and populate queues, transactions and caches in memory + JournalLoadInformation[] journalInfo = loadJournals(); compareJournals(journalInfo); + pagingManager.resumeDepages(); + + final ServerInfo dumper = new ServerInfo(this, pagingManager); + + long dumpInfoInterval = configuration.getServerDumpInterval(); + + if (dumpInfoInterval > 0) + { + scheduledPool.scheduleWithFixedDelay(new Runnable() + { + public void run() + { + HornetQServerImpl.log.info(dumper.dump()); + } + }, 0, dumpInfoInterval, TimeUnit.MILLISECONDS); + } + + // Deploy the rest of the stuff + // Deploy any predefined queues if (configuration.isFileDeploymentEnabled()) { @@ -1070,48 +1493,25 @@ } // We need to call this here, this gives any dependent server a chance to deploy its own addresses - // this needs to be done before clustering is initialised + // this needs to be done before clustering is fully activated callActivateCallbacks(); // Deply any pre-defined diverts deployDiverts(); - // This can't be created until node id is set - clusterManager = new ClusterManagerImpl(executorFactory, - this, - postOffice, - scheduledPool, - managementService, - configuration, - uuid, - configuration.isBackup(), - configuration.isClustered()); - - clusterManager.start(); - if (deploymentManager != null) { deploymentManager.start(); } - pagingManager.resumeDepages(); + clusterManager.start(); - final ServerInfo dumper = new ServerInfo(this, pagingManager); + initialised = true; - long dumpInfoInterval = configuration.getServerDumpInterval(); + // We do this at the end - we don't want things like MDBs or other connections connecting to a backup server until + // it is activated - if (dumpInfoInterval > 0) - { - scheduledPool.scheduleWithFixedDelay(new Runnable() - { - public void run() - { - HornetQServerImpl.log.info(dumper.dump()); - } - }, 0, dumpInfoInterval, TimeUnit.MILLISECONDS); - } - - initialised = true; + remotingService.start(); } /** @@ -1165,9 +1565,6 @@ recoverStoredConfigs(); - // Set the node id - must be before we load the queues into the postoffice, but after we load the journal - setNodeID(); - Map<Long, Queue> queues = new HashMap<Long, Queue>(); for (QueueBindingInfo queueBindingInfo : queueBindingInfos) @@ -1251,42 +1648,6 @@ } } - private void setNodeID() throws Exception - { - if (!configuration.isBackup()) - { - if (uuid == null) - { - uuid = storageManager.getPersistentID(); - - if (uuid == null) - { - uuid = UUIDGenerator.getInstance().generateUUID(); - - storageManager.setPersistentID(uuid); - } - - nodeID = new SimpleString(uuid.toString()); - } - } - else - { - UUID currentUUID = storageManager.getPersistentID(); - - if (currentUUID != null) - { - if (!currentUUID.equals(uuid)) - { - throw new IllegalStateException("Backup server already has an id but it's not the same as live"); - } - } - else - { - storageManager.setPersistentID(uuid); - } - } - } - private Queue createQueue(final SimpleString address, final SimpleString queueName, final SimpleString filterString, @@ -1445,19 +1806,18 @@ throw new IllegalArgumentException("Error instantiating class \"" + className + "\"", e); } } - + private static ClassLoader getThisClassLoader() { return AccessController.doPrivileged(new PrivilegedAction<ClassLoader>() - { - public ClassLoader run() - { - return ClientSessionFactoryImpl.class.getClassLoader(); - } - }); - + { + public ClassLoader run() + { + return ClientSessionFactoryImpl.class.getClassLoader(); + } + }); + } - // Inner classes // -------------------------------------------------------------------------------- Modified: branches/2_2_0_HA_Improvements/src/main/org/hornetq/utils/UUID.java =================================================================== --- branches/2_2_0_HA_Improvements/src/main/org/hornetq/utils/UUID.java 2010-06-11 04:11:42 UTC (rev 9306) +++ branches/2_2_0_HA_Improvements/src/main/org/hornetq/utils/UUID.java 2010-06-11 15:33:36 UTC (rev 9307) @@ -211,7 +211,7 @@ } return mDesc; } - + /** * Checking equality of UUIDs is easy; just compare the 128-bit number. */ Modified: branches/2_2_0_HA_Improvements/tests/src/org/hornetq/tests/unit/core/paging/impl/PagingStoreImplTest.java =================================================================== --- branches/2_2_0_HA_Improvements/tests/src/org/hornetq/tests/unit/core/paging/impl/PagingStoreImplTest.java 2010-06-11 04:11:42 UTC (rev 9306) +++ branches/2_2_0_HA_Improvements/tests/src/org/hornetq/tests/unit/core/paging/impl/PagingStoreImplTest.java 2010-06-11 15:33:36 UTC (rev 9307) @@ -984,13 +984,6 @@ return 0; } - /* (non-Javadoc) - * @see org.hornetq.core.persistence.StorageManager#getPersistentID() - */ - public UUID getPersistentID() - { - return null; - } /* (non-Javadoc) * @see org.hornetq.core.persistence.StorageManager#loadBindingJournal(java.util.List) @@ -1042,13 +1035,6 @@ } /* (non-Javadoc) - * @see org.hornetq.core.persistence.StorageManager#setPersistentID(org.hornetq.utils.UUID) - */ - public void setPersistentID(final UUID id) throws Exception - { - } - - /* (non-Javadoc) * @see org.hornetq.core.persistence.StorageManager#storeAcknowledge(long, long) */ public void storeAcknowledge(final long queueID, final long messageID) throws Exception