Merge branch 'main' into support/3.2

code/zato-server/src/zato/server/file_transfer/api.py

@@ -1,7 +1,7 @@
 # -*- coding: utf-8 -*-
 
 """
-Copyright (C) 2022, Zato Source s.r.o. https://zato.io
+Copyright (C) 2024, Zato Source s.r.o. https://zato.io
 
 Licensed under AGPLv3, see LICENSE.txt for terms and conditions.
 """
@@ -30,13 +30,15 @@
 from zato.common.typing_ import cast_
 from zato.common.util.api import new_cid, spawn_greenlet
 from zato.common.util.platform_ import is_linux, is_non_linux
+from zato.server.file_transfer.common import source_type_ftp, source_type_local, source_type_sftp, \
+    source_type_to_snapshot_maker_class
 from zato.server.file_transfer.event import FileTransferEventHandler, singleton
 from zato.server.file_transfer.observer.base import BackgroundPathInspector, PathCreatedEvent
 from zato.server.file_transfer.observer.local_ import LocalObserver
 from zato.server.file_transfer.observer.ftp import FTPObserver
 from zato.server.file_transfer.observer.sftp import SFTPObserver
-from zato.server.file_transfer.snapshot import FTPSnapshotMaker, LocalSnapshotMaker, SFTPSnapshotMaker
 
+# ################################################################################################################################
 # ################################################################################################################################
 
 if 0:
@@ -50,32 +52,19 @@
     from zato.server.file_transfer.snapshot import BaseRemoteSnapshotMaker
 
 # ################################################################################################################################
+# ################################################################################################################################
 
 logger = logging.getLogger(__name__)
 
 # ################################################################################################################################
-
-source_type_ftp   = FILE_TRANSFER.SOURCE_TYPE.FTP.id
-source_type_local = FILE_TRANSFER.SOURCE_TYPE.LOCAL.id
-source_type_sftp  = FILE_TRANSFER.SOURCE_TYPE.SFTP.id
+# ################################################################################################################################
 
 source_type_to_observer_class = {
     source_type_ftp:   FTPObserver,
     source_type_local: LocalObserver,
     source_type_sftp:  SFTPObserver,
 }
 
-source_type_to_config = {
-    source_type_ftp:  'out_ftp',
-    source_type_sftp: 'out_sftp',
-}
-
-source_type_to_snapshot_maker_class = {
-    source_type_ftp:   FTPSnapshotMaker,
-    source_type_local: LocalSnapshotMaker,
-    source_type_sftp:  SFTPSnapshotMaker,
-}
-
 # ################################################################################################################################
 # ################################################################################################################################
 
@@ -468,7 +457,7 @@ def _run_linux_inotify_loop(self) -> 'None':
                         src_path = os.path.normpath(os.path.join(dir_name, event.name))
 
                         # Get a list of all observer objects interested in that file ..
-                        observer_list = self.inotify_path_to_observer_list[dir_name] # type: list
+                        observer_list:'anylist' = self.inotify_path_to_observer_list[dir_name]
 
                         # .. and notify each one.
                         for observer in observer_list: # type: LocalObserver
@@ -479,7 +468,7 @@ def _run_linux_inotify_loop(self) -> 'None':
             except Exception:
                 logger.warning('Exception in inotify.read() `%s`', format_exc())
             finally:
-                sleep(0.25)
+                sleep(0.25) # type: ignore
 
 # ################################################################################################################################
 
@@ -493,7 +482,7 @@ def _run(self, name:'str'='', log_after_started:'bool'=False) -> 'None':
         for observer in self.observer_list: # type: LocalObserver
 
             for path in observer.path_list: # type: str
-                observer_list = self.inotify_path_to_observer_list.setdefault(path, []) # type: list
+                observer_list:'anylist' = self.inotify_path_to_observer_list.setdefault(path, [])
                 observer_list.append(observer)
 
         # Maps missing paths to all the observers interested in it.
@@ -517,7 +506,7 @@ def _run(self, name:'str'='', log_after_started:'bool'=False) -> 'None':
                 # it will add start the observer itself.
                 for path in observer.path_list:
                     if not observer.is_path_valid(path):
-                        path_observer_list = missing_path_to_inspector.setdefault(path, []) # type: list
+                        path_observer_list:'anylist' = missing_path_to_inspector.setdefault(path, [])
                         path_observer_list.append(BackgroundPathInspector(path, observer, self.observer_start_args))
 
                 # Inotify-based observers are set up here but their main loop is in _run_linux_inotify_loop ..
@@ -561,7 +550,7 @@ def get_inspector_list_by_path(self, path:'str') -> 'anydict':
             if path in observer.path_list:
 
                 # .. it was, so we append an inspector for the path, pointing to current observer.
-                path_observer_list = path_to_inspector.setdefault(path, []) # type: list
+                path_observer_list:'anylist' = path_to_inspector.setdefault(path, [])
                 path_observer_list.append(BackgroundPathInspector(path, observer, self.observer_start_args))
 
         return path_to_inspector

code/zato-server/src/zato/server/file_transfer/common.py

@@ -0,0 +1,27 @@
+# -*- coding: utf-8 -*-
+
+"""
+Copyright (C) 2024, Zato Source s.r.o. https://zato.io
+
+Licensed under AGPLv3, see LICENSE.txt for terms and conditions.
+"""
+
+# Zato
+from zato.common.api import FILE_TRANSFER
+from zato.server.file_transfer.snapshot import FTPSnapshotMaker, LocalSnapshotMaker, SFTPSnapshotMaker
+
+# ################################################################################################################################
+# ################################################################################################################################
+
+source_type_ftp   = FILE_TRANSFER.SOURCE_TYPE.FTP.id
+source_type_local = FILE_TRANSFER.SOURCE_TYPE.LOCAL.id
+source_type_sftp  = FILE_TRANSFER.SOURCE_TYPE.SFTP.id
+
+source_type_to_snapshot_maker_class = {
+    source_type_ftp:   FTPSnapshotMaker,
+    source_type_local: LocalSnapshotMaker,
+    source_type_sftp:  SFTPSnapshotMaker,
+}
+
+# ################################################################################################################################
+# ################################################################################################################################

code/zato-server/src/zato/server/file_transfer/observer/base.py

@@ -23,6 +23,7 @@
 from zato.common.api import FILE_TRANSFER
 from zato.common.util.api import spawn_greenlet
 from zato.common.util.file_transfer import path_string_list_to_list
+from zato.server.file_transfer.common import source_type_to_snapshot_maker_class
 from zato.server.file_transfer.snapshot import default_interval, DirSnapshotDiff
 
 # ################################################################################################################################
@@ -110,14 +111,18 @@ def stop(self, needs_log:'bool'=True) -> 'None':
 
     def _start(self, observer_start_args:'any_') -> 'None':
 
+        snapshot_maker = source_type_to_snapshot_maker_class[self.source_type]
+        snapshot_maker = cast_('BaseRemoteSnapshotMaker', snapshot_maker)
+        snapshot_maker.connect()
+
         for path in self.path_list:
 
             # Start only for paths that are valid - all invalid ones
             # are handled by a background path inspector.
             if self.is_path_valid(path):
                 logger.info('Starting %s file observer `%s` for `%s` (%s)',
                     self.observer_type_name, path, self.name, self.observer_type_impl)
-                _ = spawn_greenlet(self._observe_func, path, observer_start_args)
+                _ = spawn_greenlet(self._observe_func, snapshot_maker, path, maxsize, True, observer_start_args)
             else:
                 logger.info('Skipping invalid path `%s` for `%s` (%s)', path, self.name, self.observer_type_impl)