timoninmaxim commented on a change in pull request #8206:
URL: https://github.com/apache/ignite/pull/8206#discussion_r488674198
##########
File path:
modules/core/src/main/java/org/apache/ignite/internal/client/thin/ReliableChannel.java
##########
@@ -114,46 +126,32 @@
Function<ClientChannelConfiguration, ClientChannel> chFactory,
ClientConfiguration clientCfg,
IgniteBinary binary
- ) throws ClientException {
+ ) {
if (chFactory == null)
throw new NullPointerException("chFactory");
if (clientCfg == null)
throw new NullPointerException("clientCfg");
+ this.clientCfg = clientCfg;
this.chFactory = chFactory;
- List<InetSocketAddress> addrs =
parseAddresses(clientCfg.getAddresses());
-
- channels = new ClientChannelHolder[addrs.size()];
-
- for (int i = 0; i < channels.length; i++)
- channels[i] = new ClientChannelHolder(new
ClientChannelConfiguration(clientCfg, addrs.get(i)));
+ partitionAwarenessEnabled = clientCfg.isPartitionAwarenessEnabled();
- curChIdx = new Random().nextInt(channels.length); // We already
verified there is at least one address.
-
- partitionAwarenessEnabled = clientCfg.isPartitionAwarenessEnabled() &&
channels.length > 1;
-
- affinityCtx = new ClientCacheAffinityContext(binary);
-
- ClientConnectionException lastEx = null;
-
- for (int i = 0; i < channels.length; i++) {
- try {
- channels[curChIdx].getOrCreateChannel();
-
- if (partitionAwarenessEnabled)
- initAllChannelsAsync();
-
- return;
- } catch (ClientConnectionException e) {
- lastEx = e;
-
- rollCurrentChannel();
- }
- }
+ affCtx = new ClientCacheAffinityContext(binary);
+ }
- throw lastEx;
+ /**
+ * Establishing connections to servers. If partition awareness feature is
enabled connections are created
+ * for every configured server. Otherwise only default channel is
connected.
+ */
+ void initConnection() {
+ channelsInit(false);
+ if (!partitionAwarenessEnabled)
+ applyOnDefaultChannel(channel -> {
+ // do nothing, just trigger channel connection.
Review comment:
Fixed.
##########
File path:
modules/core/src/main/java/org/apache/ignite/internal/client/thin/ReliableChannel.java
##########
@@ -356,7 +318,7 @@ private boolean affinityInfoIsUpToDate(int cacheId) {
/**
* @return host:port_range address lines parsed as {@link
InetSocketAddress}.
*/
- private static List<InetSocketAddress> parseAddresses(String[] addrs)
throws ClientException {
+ private Set<InetSocketAddress> parseAddresses(String[] addrs) throws
ClientException {
Review comment:
Fixed.
##########
File path:
modules/core/src/main/java/org/apache/ignite/internal/client/thin/ReliableChannel.java
##########
@@ -473,6 +415,196 @@ public void addChannelFailListener(Runnable chFailLsnr) {
chFailLsnrs.add(chFailLsnr);
}
+ /** Should the channel initialization be stopped. */
+ private boolean stopInitCondition() {
+ return scheduledChannelsReinit.get() || closed;
+ }
+
+ /**
+ * Init channel holders to all nodes.
+ * @param force enable to replace existing channels with new holders.
+ */
+ private synchronized void initChannelHolders(boolean force) {
+ // enable parallel threads to schedule new init of channel holders
+ scheduledChannelsReinit.set(false);
+
+ if (!force && channels.get() != null)
+ return;
+
+ Set<InetSocketAddress> resolvedAddrs =
parseAddresses(clientCfg.getAddresses());
Review comment:
Fixed.
##########
File path:
modules/core/src/main/java/org/apache/ignite/internal/client/thin/ReliableChannel.java
##########
@@ -473,6 +415,196 @@ public void addChannelFailListener(Runnable chFailLsnr) {
chFailLsnrs.add(chFailLsnr);
}
+ /** Should the channel initialization be stopped. */
+ private boolean stopInitCondition() {
+ return scheduledChannelsReinit.get() || closed;
+ }
+
+ /**
+ * Init channel holders to all nodes.
+ * @param force enable to replace existing channels with new holders.
+ */
+ private synchronized void initChannelHolders(boolean force) {
+ // enable parallel threads to schedule new init of channel holders
+ scheduledChannelsReinit.set(false);
+
+ if (!force && channels.get() != null)
+ return;
+
+ Set<InetSocketAddress> resolvedAddrs =
parseAddresses(clientCfg.getAddresses());
+
+ List<ClientChannelHolder> holders =
Optional.ofNullable(channels.get()).orElse(new ArrayList<>());
Review comment:
FIxed.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
