<commit_before>package de.oneos.eventselection.amqp

import groovy.json.*
import org.apache.commons.logging.*

import com.rabbitmq.client.*
import rx.Subscription

import static de.oneos.eventselection.amqp.AMQPConstants.*
import de.oneos.eventstore.*


class AMQPEventSupplier implements EventSupplier {
    static Log log = LogFactory.getLog(AMQPEventSupplier)

    Channel channel
    Collection<EventConsumer> eventConsumers = []

    AMQPEventSupplier(Connection connection) {
        this.channel = connection.createChannel()
    }

    static String routingKey(Map<String, ?> criteria) {
        criteria.subMap(['applicationName', 'boundedContextName', 'aggregateName', 'eventName']).values().collect {
            it ?: '*'
        }.join('.')
    }

    void setEventConsumers(Collection<EventConsumer> eventConsumers) {
        this.eventConsumers.clear()
        eventConsumers.each { subscribeTo(it) }
    }

    @Override
    EventSupplier rightShift(EventConsumer eventConsumer) {
        subscribeTo(eventConsumer)
        return this
    }

    @Override
    void subscribeTo(EventConsumer eventConsumer) {
        subscribeTo(eventConsumer.eventCriteria, eventConsumer)
    }

    @Override
    void subscribeTo(Map<String, ?> criteria, EventConsumer eventConsumer) {
        deliverEvents(criteria) { EventEnvelope eventEnvelope ->
            eventConsumer.process(eventEnvelope)
            log.debug "Delivered $eventEnvelope to $eventConsumer"
        }

        eventConsumers << eventConsumer
        eventConsumer.wasRegisteredAt(this)
    }


    @SuppressWarnings("GroovyAssignabilityCheck")
    // TODO Pull up
    rx.Observable<EventEnvelope> observe(Map<String, ?> criteria) {
        return rx.Observable.create({ rx.Observer<EventEnvelope> observer ->
            deliverEvents criteria, observer.&onNext

            withEventEnvelopes criteria, observer.&onNext

            return new Subscription() {
                @Override
                void unsubscribe() {
                    // TODO implement
                }
            }
        })
    }
<commit_msg>Ignore exceptions thrown by subscribing EventConsumers in AMQPEventSupplier<commit_after>package de.oneos.eventselection.amqp

import groovy.json.*
import org.apache.commons.logging.*

import com.rabbitmq.client.*
import rx.Subscription

import static de.oneos.eventselection.amqp.AMQPConstants.*
import de.oneos.eventstore.*


class AMQPEventSupplier implements EventSupplier {
    static Log log = LogFactory.getLog(AMQPEventSupplier)

    Channel channel
    Collection<EventConsumer> eventConsumers = []

    AMQPEventSupplier(Connection connection) {
        this.channel = connection.createChannel()
    }

    static String routingKey(Map<String, ?> criteria) {
        criteria.subMap(['applicationName', 'boundedContextName', 'aggregateName', 'eventName']).values().collect {
            it ?: '*'
        }.join('.')
    }

    void setEventConsumers(Collection<EventConsumer> eventConsumers) {
        this.eventConsumers.clear()
        eventConsumers.each { subscribeTo(it) }
    }

    @Override
    EventSupplier rightShift(EventConsumer eventConsumer) {
        subscribeTo(eventConsumer)
        return this
    }

    @Override
    void subscribeTo(EventConsumer eventConsumer) {
        subscribeTo(eventConsumer.eventCriteria, eventConsumer)
    }

    @Override
    void subscribeTo(Map<String, ?> criteria, EventConsumer eventConsumer) {
        deliverEvents(criteria) { EventEnvelope eventEnvelope ->
            eventConsumer.process(eventEnvelope)
            log.debug "Delivered $eventEnvelope to $eventConsumer"
        }

        eventConsumers << eventConsumer
        try {
            eventConsumer.wasRegisteredAt(this)
        } catch(e) {
            eventConsumers.remove(eventConsumer)
            log.warn "'$e.message' during subscription of $eventConsumer", e
        }
    }


    @SuppressWarnings("GroovyAssignabilityCheck")
    // TODO Pull up
    rx.Observable<EventEnvelope> observe(Map<String, ?> criteria) {
        return rx.Observable.create({ rx.Observer<EventEnvelope> observer ->
            deliverEvents criteria, observer.&onNext

            withEventEnvelopes criteria, observer.&onNext

            return new Subscription() {
                @Override
                void unsubscribe() {
                    // TODO implement
                }
            }
        })
    }

{'fraction_non_alphanumeric': 0.09195402298850575, 'fraction_numerical': 0.0, 'mean_word_length': 2.7138075313807533, 'pattern_counts': {'":': 0, '<': 21, '<?xml version=': 0, '>': 21, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6004334', 'n_tokens_mistral': 1212, 'n_tokens_neox': 1154, 'n_words': 223}