IGNITE-14167 Simplify reconnecting, fix affinity topology change detection

This closes #16

Author: ivandasch

.travis.yml

@@ -13,6 +13,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+language: python
 sudo: required
 
 addons:
@@ -21,18 +22,27 @@ addons:
       - openjdk-8-jdk
 
 env:
-  - IGNITE_VERSION=2.9.1 IGNITE_HOME=/opt/ignite
+  global:
+    - IGNITE_VERSION=2.9.1
+    - IGNITE_HOME=/opt/ignite
 
 before_install:
   - curl -L https://apache-mirror.rbc.ru/pub/apache/ignite/${IGNITE_VERSION}/apache-ignite-slim-${IGNITE_VERSION}-bin.zip > ignite.zip
   - unzip ignite.zip -d /opt
   - mv /opt/apache-ignite-slim-${IGNITE_VERSION}-bin /opt/ignite
   - mv /opt/ignite/libs/optional/ignite-log4j2 /opt/ignite/libs/
 
-language: python
-python:
-  - "3.6"
-  - "3.7"
-  - "3.8"
-install: pip install tox-travis
+jobs:
+  include:
+    - python: '3.6'
+      arch: amd64
+      env: TOXENV=py36-no-ssl,py36-ssl,py36-ssl-password
+    - python: '3.7'
+      arch: amd64
+      env: TOXENV=py37-no-ssl,py37-ssl,py37-ssl-password
+    - python: '3.8'
+      arch: amd64
+      env: TOXENV=py38-no-ssl,py38-ssl,py38-ssl-password
+
+install: pip install tox
 script: tox

pyignite/api/affinity.py

@@ -55,11 +55,13 @@
 partition_mapping = StructArray([
     ('is_applicable', Bool),
 
-    ('cache_mapping', Conditional(lambda ctx: ctx['is_applicable'] and ctx['is_applicable'].value == 1,
+    ('cache_mapping', Conditional(['is_applicable'],
+                                  lambda ctx: ctx['is_applicable'] and ctx['is_applicable'].value == 1,
                                   lambda ctx: ctx['is_applicable'],
                                   cache_mapping, empty_cache_mapping)),
 
-    ('node_mapping', Conditional(lambda ctx: ctx['is_applicable'] and ctx['is_applicable'].value == 1,
+    ('node_mapping', Conditional(['is_applicable'],
+                                 lambda ctx: ctx['is_applicable'] and ctx['is_applicable'].value == 1,
                                  lambda ctx: ctx['is_applicable'],
                                  node_mapping, empty_node_mapping)),
 ])

pyignite/cache.py

@@ -283,6 +283,10 @@ def get_best_node(
                         parts += len(p)
 
                 self.affinity['number_of_partitions'] = parts
+
+                for conn in self.client._nodes:
+                    if not conn.alive:
+                        conn.reconnect()
             else:
                 # get number of partitions
                 parts = self.affinity.get('number_of_partitions')

pyignite/client.py

@@ -182,15 +182,9 @@ def connect(self, *args):
                     if not self.partition_aware:
                         # do not try to open more nodes
                         self._current_node = i
-                    else:
-                        # take a chance to schedule the reconnection
-                        # for all the failed connections, that was probed
-                        # before this
-                        for failed_node in self._nodes[:i]:
-                            failed_node.reconnect()
 
             except connection_errors:
-                conn._fail()
+                conn.failed = True
                 if self.partition_aware:
                     # schedule the reconnection
                     conn.reconnect()