<commit_before>/*
 * Copyright (c) 2010-2019. Axon Framework
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.axonframework.axonserver.connector.event.axon;

import io.axoniq.axonserver.grpc.event.EventWithToken;
import org.axonframework.eventhandling.EventUtils;
import org.axonframework.eventhandling.GlobalSequenceTrackingToken;
import org.axonframework.eventhandling.TrackedDomainEventData;
import org.axonframework.eventhandling.TrackedEventData;
import org.axonframework.eventhandling.TrackedEventMessage;
import org.axonframework.eventhandling.TrackingEventStream;
import org.axonframework.eventhandling.TrackingToken;
import org.axonframework.serialization.SerializedType;
import org.axonframework.serialization.Serializer;
import org.axonframework.serialization.UnknownSerializedType;
import org.axonframework.serialization.upcasting.event.EventUpcaster;
import org.axonframework.serialization.upcasting.event.NoOpEventUpcaster;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Iterator;
import java.util.Optional;
import java.util.Spliterator;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.function.Supplier;
import java.util.stream.StreamSupport;

import static org.axonframework.common.ObjectUtils.getOrDefault;

/**
 * Client-side buffer of messages received from the server. Once consumed from this buffer, the client is notified
 * of a permit being consumed, potentially triggering a permit refresh, if flow control is enabled.
 * <p>
 * This class is intended for internal use. Be cautious.
 *
 * @author <NAME>
 * @author <NAME>
 */
public class EventBuffer implements TrackingEventStream {

    private static final Logger logger = LoggerFactory.getLogger(EventBuffer.class);

    private static final int DEFAULT_POLLING_TIME_MILLIS = 500;

    private final Serializer serializer;
    private final BlockingQueue<TrackedEventData<byte[]>> events;
    private final Iterator<TrackedEventMessage<?>> eventStream;
    private final long pollingTimeMillis;

    private TrackedEventData<byte[]> peekData;
    private TrackedEventMessage<?> peekEvent;
    private Consumer<EventBuffer> closeCallback;
    private volatile RuntimeException exception;
    private volatile boolean closed;
    private Consumer<Integer> consumeListener = i -> {
    };
    private volatile Consumer<SerializedType> blacklistListener;

    /**
     * Initializes an Event Buffer, passing messages through given {@code upcasterChain} and deserializing events using
     * given {@code serializer}.
     *
     * @param upcasterChain the upcasterChain to translate serialized representations before deserializing
     * @param serializer    the serializer capable of deserializing incoming messages
     */
    public EventBuffer(EventUpcaster upcasterChain, Serializer serializer) {
        this(upcasterChain, serializer, DEFAULT_POLLING_TIME_MILLIS);
    }

    /**
     * Initializes an Event Buffer, passing messages through given {@code upcasterChain} and deserializing events using
     * given {@code serializer}.
     *
     * @param upcasterChain     the upcasterChain to translate serialized representations before deserializing
     * @param serializer        the serializer capable of deserializing incoming messages
     * @param pollingTimeMillis a {@code long} defining the polling periods used to split the up the timeout used in
     *                          {@link #hasNextAvailable(int, TimeUnit)}, ensuring nobody accidentally blocks for an
     *                          extended period of time. Defaults to 500 milliseconds
     */
    public EventBuffer(EventUpcaster upcasterChain, Serializer serializer, long pollingTimeMillis) {
        this.serializer = serializer;
        this.events = new LinkedBlockingQueue<>();
        this.eventStream = EventUtils.upcastAndDeserializeTrackedEvents(
                StreamSupport.stream(new SimpleSpliterator<>(this::poll), false),
                new GrpcMetaDataAwareSerializer(serializer),
                getOrDefault(upcasterChain, NoOpEventUpcaster.INSTANCE)
        ).iterator();
        this.pollingTimeMillis = pollingTimeMillis;
    }

    private TrackedEventData<byte[]> poll() {
        TrackedEventData<byte[]> nextItem;
        if (peekData != null) {
            nextItem = peekData;
            peekData = null;
            return nextItem;
        }
        nextItem = events.poll();
        if (nextItem != null) {
            consumeListener.accept(1);
        }
        return nextItem;
    }

    private void waitForData(long deadline) throws InterruptedException {
        long now = System.currentTimeMillis();
        if (peekData == null && now < deadline) {
            peekData = events.poll(Math.min(deadline - now, pollingTimeMillis), TimeUnit.MILLISECONDS);
            if (peekData != null) {
                consumeListener.accept(1);
            }
        }
    }

    /**
     * {@inheritDoc}
     * ----
     * <p>
     * This implementation blacklists based on the payload type of the given message.<commit_msg>Add check for exceptions while polling data<commit_after>/*
 * Copyright (c) 2010-2019. Axon Framework
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.axonframework.axonserver.connector.event.axon;

import io.axoniq.axonserver.grpc.event.EventWithToken;
import org.axonframework.eventhandling.EventUtils;
import org.axonframework.eventhandling.GlobalSequenceTrackingToken;
import org.axonframework.eventhandling.TrackedDomainEventData;
import org.axonframework.eventhandling.TrackedEventData;
import org.axonframework.eventhandling.TrackedEventMessage;
import org.axonframework.eventhandling.TrackingEventStream;
import org.axonframework.eventhandling.TrackingToken;
import org.axonframework.serialization.SerializedType;
import org.axonframework.serialization.Serializer;
import org.axonframework.serialization.UnknownSerializedType;
import org.axonframework.serialization.upcasting.event.EventUpcaster;
import org.axonframework.serialization.upcasting.event.NoOpEventUpcaster;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Iterator;
import java.util.Optional;
import java.util.Spliterator;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.function.Supplier;
import java.util.stream.StreamSupport;

import static org.axonframework.common.ObjectUtils.getOrDefault;

/**
 * Client-side buffer of messages received from the server. Once consumed from this buffer, the client is notified
 * of a permit being consumed, potentially triggering a permit refresh, if flow control is enabled.
 * <p>
 * This class is intended for internal use. Be cautious.
 *
 * @author <NAME>
 * @author <NAME>
 */
public class EventBuffer implements TrackingEventStream {

    private static final Logger logger = LoggerFactory.getLogger(EventBuffer.class);

    private static final int DEFAULT_POLLING_TIME_MILLIS = 500;

    private final Serializer serializer;
    private final BlockingQueue<TrackedEventData<byte[]>> events;
    private final Iterator<TrackedEventMessage<?>> eventStream;
    private final long pollingTimeMillis;

    private TrackedEventData<byte[]> peekData;
    private TrackedEventMessage<?> peekEvent;
    private Consumer<EventBuffer> closeCallback;
    private volatile RuntimeException exception;
    private volatile boolean closed;
    private Consumer<Integer> consumeListener = i -> {
    };
    private volatile Consumer<SerializedType> blacklistListener;

    /**
     * Initializes an Event Buffer, passing messages through given {@code upcasterChain} and deserializing events using
     * given {@code serializer}.
     *
     * @param upcasterChain the upcasterChain to translate serialized representations before deserializing
     * @param serializer    the serializer capable of deserializing incoming messages
     */
    public EventBuffer(EventUpcaster upcasterChain, Serializer serializer) {
        this(upcasterChain, serializer, DEFAULT_POLLING_TIME_MILLIS);
    }

    /**
     * Initializes an Event Buffer, passing messages through given {@code upcasterChain} and deserializing events using
     * given {@code serializer}.
     *
     * @param upcasterChain     the upcasterChain to translate serialized representations before deserializing
     * @param serializer        the serializer capable of deserializing incoming messages
     * @param pollingTimeMillis a {@code long} defining the polling periods used to split the up the timeout used in
     *                          {@link #hasNextAvailable(int, TimeUnit)}, ensuring nobody accidentally blocks for an
     *                          extended period of time. Defaults to 500 milliseconds
     */
    public EventBuffer(EventUpcaster upcasterChain, Serializer serializer, long pollingTimeMillis) {
        this.serializer = serializer;
        this.events = new LinkedBlockingQueue<>();
        this.eventStream = EventUtils.upcastAndDeserializeTrackedEvents(
                StreamSupport.stream(new SimpleSpliterator<>(this::poll), false),
                new GrpcMetaDataAwareSerializer(serializer),
                getOrDefault(upcasterChain, NoOpEventUpcaster.INSTANCE)
        ).iterator();
        this.pollingTimeMillis = pollingTimeMillis;
    }

    private TrackedEventData<byte[]> poll() {
        TrackedEventData<byte[]> nextItem;
        if (peekData != null) {
            nextItem = peekData;
            peekData = null;
            return nextItem;
        }
        nextItem = events.poll();
        if (nextItem != null) {
            consumeListener.accept(1);
        }
        return nextItem;
    }

    private void waitForData(long deadline) throws InterruptedException {
        long now = System.currentTimeMillis();
        if (peekData == null && now < deadline) {
            peekData = events.poll(Math.min(deadline - now, pollingTimeMillis), TimeUnit.MILLISECONDS);
            if (peekData != null) {
                consumeListener.accept(1);
            } else {
                checkExceptionState();
            }
        }
    }

    /**
     * {@inheritDoc}
     * ----
     * <p>
     * This implementation blacklists based on the payload type of the given message.
{'fraction_non_alphanumeric': 0.07734470158343483, 'fraction_numerical': 0.003828084217852793, 'mean_word_length': 3.5815065763252294, 'pattern_counts': {'":': 0, '<': 39, '<?xml version=': 0, '>': 39, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '6726003', 'n_tokens_mistral': 3051, 'n_tokens_neox': 2906, 'n_words': 885}