Merge pull request #223 from poissoncorp/RDBC-847

RDBC-847 Implement documentStore.Maintenance.Server.ForNode

Author: poissoncorp

ravendb/documents/bulk_insert_operation.py

@@ -413,9 +413,6 @@ def _get_id(self, entity: object) -> str:
         self._generate_entity_id_on_the_client.try_set_identity(entity, key)
         return key
 
-    # todo: CountersBulkInsert
-    # todo: CountersBulkInsertOperation
-
     class TimeSeriesBulkInsertBase(ABC):
         def __init__(self, operation: Optional[BulkInsertOperation], id_: Optional[str], name: Optional[str]):
             operation._end_previous_command_if_needed()

ravendb/documents/operations/executor.py

@@ -126,7 +126,7 @@ def request_executor(self) -> RequestExecutor:
     def server(self) -> ServerOperationExecutor:
         if self._server_operation_executor is not None:
             return self._server_operation_executor
-        self._server_operation_executor = ServerOperationExecutor(self._store)
+        self._server_operation_executor = ServerOperationExecutor.from_store(self._store)
         return self._server_operation_executor
 
     def for_database(self, database_name: str) -> MaintenanceOperationExecutor:

ravendb/serverwide/server_operation_executor.py

@@ -28,48 +28,95 @@ class ConnectionStringType(enum.Enum):
 
 
 class ServerOperationExecutor:
-    def __init__(self, store: DocumentStore):
+    def __init__(
+        self,
+        store: DocumentStore,
+        request_executor: ClusterRequestExecutor,
+        initial_request_executor: ClusterRequestExecutor = None,
+        cache: CaseInsensitiveDict = None,
+        node_tag: str = None,
+    ):
         if store is None:
             raise ValueError("Store cannot be None")
-        request_executor = self.create_request_executor(store)
-
         if request_executor is None:
             raise ValueError("Request Executor cannot be None")
-        self.__store = store
-        self.__request_executor = request_executor
-        self.__initial_request_executor = None
-        self.__node_tag = None
-        self.__cache = CaseInsensitiveDict()
+        if cache is None:
+            cache = CaseInsensitiveDict()
+
+        self._store = store
+        self._request_executor = request_executor
+        self._initial_request_executor = initial_request_executor
+        self._node_tag = node_tag
+        self._cache = cache
+
+        store.register_events_for_request_executor(self._request_executor)
+
+        if self._node_tag is None:
+            self._store.add_after_close(lambda: self._request_executor.close())
+
+    @classmethod
+    def from_store(cls, store: DocumentStore):
+        return cls(store, cls.create_request_executor(store), None, CaseInsensitiveDict(), None)
+
+    def for_node(self, node_tag: str) -> ServerOperationExecutor:
+        if not node_tag or node_tag.isspace():
+            raise ValueError("Value cannot be None or whitespace")
+
+        if self._node_tag and self._node_tag.lower() == node_tag.lower():
+            return self
 
-        store.register_events_for_request_executor(self.__request_executor)
-        store.add_after_close(self.close)
+        if self._store.conventions.disable_topology_updates:
+            raise RuntimeError(
+                "Cannot switch server operation executor, because conventions.disable_topology_updates is set to 'True'"
+            )
+
+        if node_tag in self._cache:
+            return self._cache[node_tag]
+
+        request_executor = self._initial_request_executor or self._request_executor
+        topology = self._get_topology(self._request_executor)
+
+        node = next((node for node in topology.nodes if node_tag.lower() == node.cluster_tag.lower()), None)
+
+        if node is None:
+            available_nodes = str.join(", ", [node.cluster_tag for node in topology.nodes])
+            raise RuntimeError(f"Could not find node '{node_tag}' in the topology. Available nodes: {available_nodes}")
+
+        cluster_executor = ClusterRequestExecutor.create_for_single_node(
+            node.url,
+            self._store.thread_pool_executor,
+            self._store.conventions,
+            self._store.certificate_pem_path,
+            self._store.trust_store_path,
+        )
+        return ServerOperationExecutor(self._store, cluster_executor, request_executor, self._cache, node.cluster_tag)
 
     def send(self, operation: ServerOperation[_T_OperationResult]) -> Optional[_T_OperationResult]:
-        command = operation.get_command(self.__request_executor.conventions)
-        self.__request_executor.execute_command(command)
+        command = operation.get_command(self._request_executor.conventions)
+        self._request_executor.execute_command(command)
 
         if isinstance(operation, ServerOperation):
             return command.result
 
     def send_async(self, operation: ServerOperation[OperationIdResult]) -> Operation:
-        command = operation.get_command(self.__request_executor.conventions)
+        command = operation.get_command(self._request_executor.conventions)
 
-        self.__request_executor.execute_command(command)
+        self._request_executor.execute_command(command)
         return ServerWideOperation(
-            self.__request_executor,
-            self.__request_executor.conventions,
+            self._request_executor,
+            self._request_executor.conventions,
             command.result.operation_id,
             command.selected_node_tag if command.selected_node_tag else command.result.operation_node_tag,
         )
 
     def close(self) -> None:
-        if self.__node_tag is not None:
+        if self._node_tag is not None:
             return
 
-        if self.__request_executor is not None:
-            self.__request_executor.close()
+        if self._request_executor is not None:
+            self._request_executor.close()
 
-        cache = self.__cache
+        cache = self._cache
         if cache is not None:
             for key, value in cache.items():
                 request_executor = value._request_executor
@@ -78,8 +125,8 @@ def close(self) -> None:
 
             cache.clear()
 
-    def __get_topology(self, request_executor: ClusterRequestExecutor) -> Topology:
-        topology: Topology = None
+    def _get_topology(self, request_executor: ClusterRequestExecutor) -> Topology:
+        topology: Optional[Topology] = None
         try:
             topology = request_executor.topology
             if topology is None:
@@ -92,8 +139,8 @@ def __get_topology(self, request_executor: ClusterRequestExecutor) -> Topology:
 
                 topology = request_executor.topology
 
-        except:
-            pass
+        except Exception:
+            ...  # ignored
 
         if topology is None:
             raise RuntimeError("Could not fetch the topology")