def get_or_create_monitor()

in aws_advanced_python_wrapper/host_monitoring_plugin.py [0:0]

• 26 lines of code
• 8 McCabe index (conditional complexity)

    def get_or_create_monitor(self, host_aliases: FrozenSet[str], monitor_supplier: Callable) -> Monitor:
        if not host_aliases:
            raise AwsWrapperError(Messages.get("MonitoringThreadContainer.EmptyHostKeys"))

        with self._monitor_lock:
            monitor = None
            any_alias = next(iter(host_aliases))
            for host_alias in host_aliases:
                monitor = self._monitor_map.get(host_alias)
                any_alias = host_alias
                if monitor is not None:
                    break

            def _get_or_create_monitor(_) -> Monitor:
                supplied_monitor = monitor_supplier()
                if supplied_monitor is None:
                    raise AwsWrapperError(Messages.get("MonitoringThreadContainer.SupplierMonitorNone"))
                self._tasks_map.compute_if_absent(
                    supplied_monitor, lambda _: MonitoringThreadContainer._executor.submit(supplied_monitor.run))
                return supplied_monitor

            if monitor is None:
                monitor = self._monitor_map.compute_if_absent(any_alias, _get_or_create_monitor)
                if monitor is None:
                    raise AwsWrapperError(
                        Messages.get_formatted("MonitoringThreadContainer.ErrorGettingMonitor", host_aliases))

            for host_alias in host_aliases:
                self._monitor_map.put_if_absent(host_alias, monitor)

            return monitor