adctk.publisher.MultiPublisher Class Reference

Interface for a group of publishers all being fed the same message(s). More...

Public Member Functions
	__init__ (self, bool strict, str|list[str]|dict|None plugins)
str	version (self)
None	add (self, Publisher publisher)
int	publish (self, builder.Builder b)
None	terminate (self)
	finalize() all configured publishers
None	pause (self)
	pause() all configured publishers
None	resume (self)
	resume() all configured publishers
list[str]	get_names (self)

Public Attributes
	defaults
	paused
	plugin_prefix
	strict
	factory
	pvec

Protected Member Functions
	_config_from_dict (self, dict[dict[str]] plugins, strict)
	_config_from_list (self, list[str] plugins, bool strict)
	_config_from_env (self, str env_name, bool strict)

Protected Attributes
	_version

Detailed Description

Interface for a group of publishers all being fed the same message(s).

Constructor & Destructor Documentation

__init__()

adctk.publisher.MultiPublisher.__init__	(		self,
		bool	strict,
		str|list[str]|dict|None	plugins
	)

Parameters

strict	x
plugins

    def __init__(self, strict: bool, plugins: str|list[str]|dict|None):
        """! @param strict x
            @param plugins
        """
        self._version = "1.0.0"
        self.defaults = None
        self.paused = False
        self.plugin_prefix = None
        self.strict = strict
        self.pvec : [Publisher] = []
        from . import factory # break import loop
        self.factory = factory.Factory
        if plugins is None:
            return
        if isinstance(plugins, str):
            self._config_from_env(plugins, strict)
            return
        if isinstance(plugins, list):
            if len(plugins):
                self._config_from_list(plugins, strict)
            else:
                self._config_from_env("ADC_MULTI_PUBLISHER_NAMES", strict)
            return
        if isinstance(plugins, dict):
            self._config_from_dict(plugins, strict)
            return
        raise ValueError("plugins argument of unexpected type: {plugins}")
 

Member Function Documentation

_config_from_dict()

adctk.publisher.MultiPublisher._config_from_dict (   self, dict[dict[str]]  plugins,   strict  )

protected

@brief custom conf plugin X using plugins[X] dictionary.

    def _config_from_dict(self, plugins: dict[dict[str]], strict):
        """@brief custom conf plugin X using plugins[X] dictionary."""
        for plug, d in plugins.items():
            p = self.factory.get_publisher(plug)
            if strict and not p:
                raise ValueError(f"MultiPublisher unable to construct plugin {plug}")
            if d:
                if not isinstance(d, dict):
                    raise ValueError(f"plugins must be a dictionary of dictionaries of strings, not {plugins}")
                p.config(d)
            else:
                p.config({})
            p.initialize()
            self.add(p)
 

_config_from_env()

adctk.publisher.MultiPublisher._config_from_env (   self, str  env_name, bool  strict  )

protected

@brief configure with list following env_name.split(":")

    def _config_from_env(self, env_name: str, strict: bool):
        """@brief configure with list following env_name.split(":")"""
        env_val = os.getenv(env_name, "")
        if env_val is None or not len(env_val):
            return
        plugins = env_val.split(":")
        self._config_from_list(plugins, strict)
 

_config_from_list()

adctk.publisher.MultiPublisher._config_from_list (   self, list[str]  plugins, bool  strict  )

protected

@brief configure each plugin named in the list with defaults & env.

    def _config_from_list(self, plugins: list[str], strict: bool):
        """@brief configure each plugin named in the list with defaults & env."""
        for i in plugins:
            p = self.factory.get_publisher(i)
            if strict and not p:
                raise ValueError(f"MultiPublisher unable to construct plugin {i}")
            self.add(p)
 

add()

None adctk.publisher.MultiPublisher.add	(		self,
		Publisher	publisher
	)

 @param Add publisher to the dispatch list

    def add(self, publisher: Publisher) -> None:
        """ @param Add publisher to the dispatch list"""
        if not publisher:
            return None
        self.pvec.append(publisher)
        print(f"DEB: Adding publisher {publisher.name()}")
        return None
 

get_names()

list[str] adctk.publisher.MultiPublisher.get_names

(

self

)

    def get_names(self) -> list[str]:
        n = []
        for i in self.pvec:
            n.append(i.name())
        return n

pause()

None adctk.publisher.MultiPublisher.pause

(

self

)

pause() all configured publishers

    def pause(self) -> None:
        """! pause() all configured publishers"""
        # print(f"PAUSING {len(self.pvec)} PUBLISHERS")
        for pub in self.pvec:
            pub.pause()
        self.paused = True
 

publish()

int adctk.publisher.MultiPublisher.publish	(		self,
		builder.Builder	b
	)

Parameters

b	builder to publish via all configured publishers.

Returns

count of publishers for which publish() returns an error.

    def publish(self, b: builder.Builder) -> int:
        """! @param b builder to publish via all configured publishers.
 
            @return count of publishers for which publish() returns an error.
        """
        if self.paused:
            return 0
        err = 0
        for pub in self.pvec:
            e = pub.publish(b)
            if e:
                err += 1
                log.logger.debug("Publish failed for plugin %s", pub.name())
        return err
 

resume()

None adctk.publisher.MultiPublisher.resume

(

self

)

resume() all configured publishers

    def resume(self) -> None:
        """! resume() all configured publishers"""
        # print(f"RESUMING {len(self.pvec)} PUBLISHERS")
        for pub in self.pvec:
            pub.resume()
        self.paused = False
 

terminate()

None adctk.publisher.MultiPublisher.terminate

(

self

)

finalize() all configured publishers

    def terminate(self) -> None:
        """! finalize() all configured publishers"""
        for pub in self.pvec:
            pub.finalize()
        self.pvec = []
 

version()

str adctk.publisher.MultiPublisher.version

(

self

)

 @return the version of the MultiPublisher

    def version(self) -> str:
        """ @return the version of the MultiPublisher"""
        return self._version
 

Member Data Documentation

_version

adctk.publisher.MultiPublisher._version

protected

defaults

adctk.publisher.MultiPublisher.defaults

factory

adctk.publisher.MultiPublisher.factory

paused

adctk.publisher.MultiPublisher.paused

plugin_prefix

adctk.publisher.MultiPublisher.plugin_prefix

pvec

adctk.publisher.MultiPublisher.pvec

strict

adctk.publisher.MultiPublisher.strict

---

The documentation for this class was generated from the following file:

• src/adctk/publisher.py