class DummySource(Source):
    def stop(self):
        pass

    def start(self):
        pass class Pipeline:
    def __init__(self, builder, plumber):
        self.id = builder.id
        self.plumber = plumber
        self.status = Status.stopped
        self.auto_start = builder.auto_start
        self.source = builder.source_class(plumber, builder.source_params)
        self.transient_previous = None
        self.wiretap = None
        for destination in builder.to_list:
            if destination[0] == Wiretap:
                self.wiretap = destination[1]
                continue
            channel = Channel(self.plumber, {"wiretap": self.wiretap})
            if self.source.chain is None:
                self.source.chain = channel
            if issubclass(destination[0], Destination):
                destination_obj = destination[0](plumber, destination[1])
                processor = Processor(destination_obj)
            if issubclass(destination[0], Splitter):
                destination_obj = destination[0](plumber)
                processor = SplitterProcessor(destination_obj)
            if issubclass(destination[0], Filter):
                destination_obj = destination[0](plumber)
                processor = FilterProcessor(destination_obj)
            if issubclass(destination[0], Aggregator):
                destination_obj = destination[0](plumber, destination[1])
                processor = AggregatorProcessor(destination_obj)
            if issubclass(destination[0], RoutingSlip):
                destination_obj = destination[0](plumber, destination[1])
                processor = RoutingSlipProcessor(destination_obj)
            if issubclass(destination[0], DynamicRouter):
                destination_obj = destination[0](plumber, destination[1])
                processor = DynamicRouterProcessor(destination_obj)
            if issubclass(destination[0], Validator):
                destination_obj = destination[0](plumber)
                processor = ValidatorProcessor(destination_obj)
            if destination[0] == Multicast:
                destination_obj = destination[0](plumber, destination[1])
                processor = MulticastProcessor(destination_obj)
            if destination[0] == ContentBasedRouter:
                destination_obj = destination[0](plumber, destination[1])
                processor = ContentBasedRouterProcessor(destination_obj)
            if destination[0] == Resequencer:
                destination_obj = destination[0](plumber, destination[1])
                processor = ResequencerProcessor(destination_obj)
            channel.next = processor
            if self.transient_previous is not None:
                self.transient_previous.next = channel
            self.transient_previous = processor
            self.wiretap = None

    def start(self):
        self.status = Status.starting
        self.source.start()
        self.status = Status.running

    def stop(self):
        self.status = Status.stopping
        self.source.stop()
        self.status = Status.stopped class Source:
    def __init__(self, plumber, params):
        self.plumber = plumber
        self.chain = None
        self.params = params

    def start(self):
        raise NotImplementedError("Sources should implement their start method")

    def stop(self):
        raise NotImplementedError("Sources should implement their stop method") class Destination:
    def __init__(self, plumber, params):
        self.plumber = plumber
        self.params = params

    def process(self, exchange):
        raise NotImplementedError("Subclass of destination needs to implement process method") class PipelineBuilder:
    def __init__(self):
        self.source_class = None
        self.source_params = None
        self.to_list = []
        self.id = None
        self.auto_start = True

    def build(self):
        raise NotImplementedError

    def build_with_plumber(self, plumber):
        raise NotImplementedError class Splitter:
    def __init__(self, plumber):
        self.plumber = plumber

    def split(self, exchange):
        raise NotImplementedError("Derived classes should implement the split method") class Filter:
    def __init__(self, plumber):
        self.plumber = plumber

    def filter(self, exchange):
        raise NotImplementedError("Derived classes should implement the filter method") class Multicast:
    def __init__(self, plumber, params):
        self.plumber = plumber
        self.params = params
        self.aggregator = params["aggregator"](plumber, {})
        self.pipelines = []
        for builder in params["pipelines"]:
            self.pipelines.append(builder.build_with_plumber(plumber)) class Aggregator:
    def __init__(self, plumber, params):
        self.plumber = plumber
        if "timeout" in params:
            self.timeout = params["timeout"]
        else:
            self.timeout = None
        if "count" in params:
            self.count = params["count"]
        else:
            self.count = None

    def aggregate(self, old_exchange, current_exchange):
        return current_exchange class ContentBasedRouter:
    def __init__(self, plumber, params):
        self.branches = []
        self.plumber = plumber
        for condition, builder in params["branches"]:
            self.branches.append((condition, builder.build_with_plumber(plumber)))

    def get_valid_pipeline(self, exchange):
        for condition, pipeline in self.branches:
            valid = condition(exchange)
            if valid:
                return pipeline
        return None class RoutingSlip:
    def __init__(self, plumber, params):
        self.plumber = plumber
        self.params = params

    def slip(self, exchange):
        raise NotImplementedError("Subclass must implement the slip method") class DynamicRouter:
    def __init__(self, plumber, params):
        self.plumber = plumber
        self.params = params

    def route(self, exchange):
        raise NotImplementedError("Subclass must implement the route method") class Resequencer:
    def __init__(self, plumber, params):
        self.plumber = plumber
        if "timeout" in params:
            self.timeout = params["timeout"]
        else:
            self.timeout = None
        if "count" in params:
            self.count = params["count"]
        else:
            self.count = None
        if "key_extractor" in params:
            self.key_extractor = params["key_extractor"]
        else:
            self.key_extractor = None
        if "reverse" in params:
            self.reverse = params["reverse"]
        else:
            self.reverse = False

    def resequence(self, exchanges):
        exchanges.sort(key=self.key_extractor, reverse=self.reverse) class Validator:
    def __init__(self, plumber):
        self.plumber = plumber

    def validate(self, exchange):
        raise NotImplementedError("Derived classes should implement the filter method") class Wiretap:
    def __init__(self, endpoint):
        self.endpoint = endpoint
from pypipeline.core.DummySource import DummySource
from .Pipeline import Pipeline
from .Source import Source
from .Destination import Destination
from .PipelineBuilder import PipelineBuilder
from pypipeline.eip.split.Splitter import Splitter
from pypipeline.eip.filter.Filter import Filter
from pypipeline.eip.multicast.Multicast import Multicast
from pypipeline.eip.aggregate.Aggregator import Aggregator
from pypipeline.eip.cbr.ContentBasedRouter import ContentBasedRouter
from pypipeline.eip.routing_slip.RoutingSlip import RoutingSlip
from pypipeline.eip.dynamic_router.DynamicRouter import DynamicRouter
from pypipeline.eip.resequence.Resequencer import Resequencer
from pypipeline.eip.validate.Validator import Validator
from pypipeline.eip.wiretap.Wiretap import Wiretap


class DslPipelineBuilder(PipelineBuilder):

    def __init__(self):
        super().__init__()
        self._destination_stack = []
        self._builder_stack = [self, ]
        self._when_condition = None

    def source(self, endpoint, params=None):
        if params is None:
            params = {}
        assert issubclass(endpoint, Source), "The source class should be a subclass of Source"
        assert self._builder_stack[-1].source_class is None, "There can only be one source in a pipeline"
        self._builder_stack[-1].source_class = endpoint
        self._builder_stack[-1].source_params = params

        assert issubclass(self._builder_stack[-1].source_class, Source), "The source class should be a subclass of Source"