<commit_before>/*
 * Copyright (c) 2011-2016 Pivotal Software Inc, All Rights Reserved.
 *
 * 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 reactor.aeron.publisher;

import java.util.Arrays;
import java.util.Iterator;
import java.util.function.Consumer;

import org.reactivestreams.Subscriber;
import reactor.aeron.Context;
import reactor.aeron.utils.AeronInfra;
import reactor.aeron.utils.AeronUtils;
import reactor.aeron.utils.DemandTracker;
import reactor.aeron.utils.ServiceMessagePublicationFailedException;
import reactor.aeron.utils.ServiceMessageType;
import reactor.aeron.utils.SignalType;
import reactor.core.flow.MultiProducer;
import reactor.core.flow.Producer;
import reactor.core.state.Cancellable;
import reactor.core.state.Completable;
import reactor.core.state.Introspectable;
import reactor.core.state.Requestable;
import reactor.core.util.BackpressureUtils;
import reactor.core.util.Exceptions;
import reactor.core.util.Logger;
import reactor.io.buffer.Buffer;
import uk.co.real_logic.aeron.FragmentAssembler;
import uk.co.real_logic.aeron.Subscription;
import uk.co.real_logic.aeron.logbuffer.FragmentHandler;
import uk.co.real_logic.aeron.logbuffer.Header;
import uk.co.real_logic.agrona.DirectBuffer;
import uk.co.real_logic.agrona.concurrent.IdleStrategy;

/**
 * Signals receiver functionality which polls for signals sent by senders
 */
class SignalPoller implements org.reactivestreams.Subscription, Runnable, Producer,
                                     Requestable, Completable,
                                     Cancellable, MultiProducer, Introspectable {

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

	private final AeronInfra aeronInfra;

	private final Subscriber<? super Buffer> subscriber;

	private final Consumer<Boolean> shutdownTask;

	private final DemandTracker demandTracker;

	private final Context context;

	private final SignalFragmentHandler signalFragmentHandler;

	private volatile boolean running;

	private volatile uk.co.real_logic.aeron.Subscription signalSub;

	private final ServiceMessageSender serviceMessageSender;

	private abstract class BaseFragmentHandler implements FragmentHandler {

		private final FragmentAssembler fragmentAssembler = new FragmentAssembler(new FragmentHandler() {
			@Override
			public void onFragment(DirectBuffer buffer, int offset, int length, Header header) {
				doOnFragment(buffer, offset, length, header);
			}
		});

		@Override
		public void onFragment(DirectBuffer buffer, int offset, int length, Header header) {
			fragmentAssembler.onFragment(buffer, offset, length, header);
		}

		private void doOnFragment(DirectBuffer buffer, int offset, int length, Header header) {
			byte[] data = new byte[length - 1];
			buffer.getBytes(offset + 1, data);
			byte signalTypeCode = buffer.getByte(offset);
			try {
				if (!handleSignal(signalTypeCode, data, header.sessionId())) {
					logger.error("Message with unknown signal type code of {} and length of {} was ignored",
							signalTypeCode, data.length);
				}
			} catch (Throwable t) {
				Exceptions.throwIfFatal(t);
				subscriber.onError(t);
			}
		}

		/**
		 * Handles signal with type code of <code>signalTypeCode</code> and
		 * content of <code>data</code>
		 *
		 * @param signalTypeCode signal type code
		 * @param data signal data
		 * @param sessionId Aeron sessionId
		 * @return true if signal was handled and false otherwise
		 */
		abstract boolean handleSignal(byte signalTypeCode, byte[] data, int sessionId);

	}

	/**
	 * Handler for Next, Complete and Error signals
	 */
	private class SignalFragmentHandler extends BaseFragmentHandler {

		/**
		 * If should read a single message from Aeron.
		 * Used to check if Complete signal was sent before any events were
		 * requested via a subscription
		 */
		boolean checkForComplete;

		/**
		 * Message read from Aeron but not yet forwarded into a subscriber
		 */
		Buffer reservedNextSignal;

		/**
		 * Number of Next signals received
		 */
		int nextSignalCounter;

		/**
		 * Complete signal was received from one of senders
		 */
		private boolean isTerminalSignalReceived = false;

		@Override
		boolean handleSignal(byte signalTypeCode, byte[] data, int sessionId) {
			if (signalTypeCode == SignalType.Next.getCode()) {
				Buffer buffer = Buffer.wrap(data);
				if (checkForComplete) {
					reservedNextSignal = buffer;
				} else {
					subscriber.onNext(buffer);
					nextSignalCounter++;
				}
			} else if (signalTypeCode == SignalType.Complete.getCode()) {
				subscriber.onComplete();

				isTerminalSignalReceived = true;
			} else if (signalTypeCode == SignalType.Error.getCode()) {
				Throwable t = context.exceptionSerializer().deserialize(data);
				subscriber.onError(t);

				isTerminalSignalReceived = true;
			} else {
				return false;
			}
			return true;
		}

		int getAndResetNextSignalCounter() {
			int result = nextSignalCounter;
			nextSignalCounter = 0;
			return result;
		}

		public boolean hasReservedNextSignal() {
			return reservedNextSignal != null;
		}

		public void processReservedNextSignal() {
			subscriber.onNext(reservedNextSignal);
			reservedNextSignal = null;
		}

		public boolean isTerminalSignalReceived() {
			return isTerminalSignalReceived;
		}
	}

	@Override
	public int getMode() {
		return INNER;
	}

	@Override
	public String getName() {
		return SignalPoller.class.getSimpleName();
	}

	public SignalPoller(Context context,
						ServiceMessageSender serviceMessageSender,
						Subscriber<? super Buffer> subscriber,
						AeronInfra aeronInfra,
						Consumer<Boolean> shutdownTask) {

		this.context = context;
		this.serviceMessageSender = serviceMessageSender;
		this.subscriber = subscriber;
		this.aeronInfra = aeronInfra;
		this.shutdownTask = shutdownTask;
		this.signalFragmentHandler = new SignalFragmentHandler();
		this.demandTracker = new DemandTracker();
	}

	@Override
	public void run() {
		running = true;
		logger.debug("Signal poller started, sessionId: {}", serviceMessageSender.getSessionId());

		uk.co.real_logic.aeron.Subscription signalSub = aeronInfra.addSubscription(context.receiverChannel(),
				context.streamId());

		this.signalSub = signalSub;

		setSubscriberSubscription();

		final IdleStrategy idleStrategy = AeronUtils.newBackoffIdleStrategy();
		final DemandTracker demandTracker = this.demandTracker;
		boolean isTerminalSignalReceived = false;
		long demand = 0;
		try {
			while (running) {
				if (demand == 0) {
					demand = demandTracker.getAndReset();
				}

				int nFragmentsReceived = 0;
				int fragmentLimit = (int) Math.min(demand, context.signalPollerFragmentLimit());
				if (fragmentLimit == 0) {
					if (!signalFragmentHandler.hasReservedNextSignal()) {
						checkForTerminalSignal(signalSub);
					}
				} else {
					if (signalFragmentHandler.hasReservedNextSignal()) {
						signalFragmentHandler.processReservedNextSignal();
						fragmentLimit--;
						demand--;
					}
					if (fragmentLimit > 0) {
						nFragmentsReceived = signalSub.poll(signalFragmentHandler, fragmentLimit);
						demand -= signalFragmentHandler.getAndResetNextSignalCounter();
					}
				}
				idleStrategy.idle(nFragmentsReceived);

				if (signalFragmentHandler.isTerminalSignalReceived()) {
					isTerminalSignalReceived = true;
					running = false;
					break;
				}
			}
		} finally {
			aeronInfra.close(signalSub);

			logger.trace("about to execute shutdownTask");
			shutdownTask.accept(isTerminalSignalReceived);
		}

		logger.debug("Signal poller shutdown, sessionId: {}", serviceMessageSender.getSessionId());
	}

	private void setSubscriberSubscription() {
		//TODO: Possible timing issue due to ServiceMessagePoller termination
		try {
			if(running) {
				subscriber.onSubscribe(this);
			}
		} catch (Throwable t) {
			Exceptions.throwIfFatal(t);
			subscriber.onError(t);
		}
	}

	private void checkForTerminalSignal(Subscription nextCompleteSub) {
		signalFragmentHandler.checkForComplete = true;
		nextCompleteSub.poll(signalFragmentHandler, 1);
		signalFragmentHandler.checkForComplete = false;
	}

	public void shutdown() {
		running = false;
	}

	@Override
	public Iterator<?> downstreams() {
		String up1 = signalSub != null ? signalSub.channel()+"/"+ signalSub.streamId() :
				context.receiverChannel()+"/"+context.streamId();

		return Arrays.asList(up1, serviceMessageSender).iterator();
	}

	@Override
	public long downstreamCount() {
		return running ? 2 : 0;
	}

	@Override
	public boolean isCancelled() {
		return !running;
	}

	@Override
	public boolean isStarted() {
		return running;
	}

	@Override
	public boolean isTerminated() {
		return !running;
	}

	@Override
	public Object downstream() {
		return subscriber;
	}

	@Override
	public long requestedFromDownstream() {
		return demandTracker.current();
	}

	@Override
	public void request(long n) {
		if (running && BackpressureUtils.checkRequest(n, subscriber)) {
			try {
				serviceMessageSender.sendRequest(n);
				demandTracker.request(n);
			} catch (Exception e) {
				subscriber.onError(new ServiceMessagePublicationFailedException(ServiceMessageType.Request, e));
			}
		}
	}

	@Override
	public void cancel() {
		running = false;
	}
}
<commit_msg>default interface for Introspectable<commit_after>/*
 * Copyright (c) 2011-2016 Pivotal Software Inc, All Rights Reserved.
 *
 * 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 reactor.aeron.publisher;

import java.util.Arrays;
import java.util.Iterator;
import java.util.function.Consumer;

import org.reactivestreams.Subscriber;
import reactor.aeron.Context;
import reactor.aeron.utils.AeronInfra;
import reactor.aeron.utils.AeronUtils;
import reactor.aeron.utils.DemandTracker;
import reactor.aeron.utils.ServiceMessagePublicationFailedException;
import reactor.aeron.utils.ServiceMessageType;
import reactor.aeron.utils.SignalType;
import reactor.core.flow.MultiProducer;
import reactor.core.flow.Producer;
import reactor.core.state.Cancellable;
import reactor.core.state.Completable;
import reactor.core.state.Introspectable;
import reactor.core.state.Requestable;
import reactor.core.util.BackpressureUtils;
import reactor.core.util.Exceptions;
import reactor.core.util.Logger;
import reactor.io.buffer.Buffer;
import uk.co.real_logic.aeron.FragmentAssembler;
import uk.co.real_logic.aeron.Subscription;
import uk.co.real_logic.aeron.logbuffer.FragmentHandler;
import uk.co.real_logic.aeron.logbuffer.Header;
import uk.co.real_logic.agrona.DirectBuffer;
import uk.co.real_logic.agrona.concurrent.IdleStrategy;

/**
 * Signals receiver functionality which polls for signals sent by senders
 */
class SignalPoller implements org.reactivestreams.Subscription, Runnable, Producer,
                                     Requestable, Completable,
                                     Cancellable, MultiProducer, Introspectable {

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

	private final AeronInfra aeronInfra;

	private final Subscriber<? super Buffer> subscriber;

	private final Consumer<Boolean> shutdownTask;

	private final DemandTracker demandTracker;

	private final Context context;

	private final SignalFragmentHandler signalFragmentHandler;

	private volatile boolean running;

	private volatile uk.co.real_logic.aeron.Subscription signalSub;

	private final ServiceMessageSender serviceMessageSender;

	private abstract class BaseFragmentHandler implements FragmentHandler {

		private final FragmentAssembler fragmentAssembler = new FragmentAssembler(new FragmentHandler() {
			@Override
			public void onFragment(DirectBuffer buffer, int offset, int length, Header header) {
				doOnFragment(buffer, offset, length, header);
			}
		});

		@Override
		public void onFragment(DirectBuffer buffer, int offset, int length, Header header) {
			fragmentAssembler.onFragment(buffer, offset, length, header);
		}

		private void doOnFragment(DirectBuffer buffer, int offset, int length, Header header) {
			byte[] data = new byte[length - 1];
			buffer.getBytes(offset + 1, data);
			byte signalTypeCode = buffer.getByte(offset);
			try {
				if (!handleSignal(signalTypeCode, data, header.sessionId())) {
					logger.error("Message with unknown signal type code of {} and length of {} was ignored",
							signalTypeCode, data.length);
				}
			} catch (Throwable t) {
				Exceptions.throwIfFatal(t);
				subscriber.onError(t);
			}
		}

		/**
		 * Handles signal with type code of <code>signalTypeCode</code> and
		 * content of <code>data</code>
		 *
		 * @param signalTypeCode signal type code
		 * @param data signal data
		 * @param sessionId Aeron sessionId
		 * @return true if signal was handled and false otherwise
		 */
		abstract boolean handleSignal(byte signalTypeCode, byte[] data, int sessionId);

	}

	/**
	 * Handler for Next, Complete and Error signals
	 */
	private class SignalFragmentHandler extends BaseFragmentHandler {

		/**
		 * If should read a single message from Aeron.
		 * Used to check if Complete signal was sent before any events were
		 * requested via a subscription
		 */
		boolean checkForComplete;

		/**
		 * Message read from Aeron but not yet forwarded into a subscriber
		 */
		Buffer reservedNextSignal;

		/**
		 * Number of Next signals received
		 */
		int nextSignalCounter;

		/**
		 * Complete signal was received from one of senders
		 */
		private boolean isTerminalSignalReceived = false;

		@Override
		boolean handleSignal(byte signalTypeCode, byte[] data, int sessionId) {
			if (signalTypeCode == SignalType.Next.getCode()) {
				Buffer buffer = Buffer.wrap(data);
				if (checkForComplete) {
					reservedNextSignal = buffer;
				} else {
					subscriber.onNext(buffer);
					nextSignalCounter++;
				}
			} else if (signalTypeCode == SignalType.Complete.getCode()) {
				subscriber.onComplete();

				isTerminalSignalReceived = true;
			} else if (signalTypeCode == SignalType.Error.getCode()) {
				Throwable t = context.exceptionSerializer().deserialize(data);
				subscriber.onError(t);

				isTerminalSignalReceived = true;
			} else {
				return false;
			}
			return true;
		}

		int getAndResetNextSignalCounter() {
			int result = nextSignalCounter;
			nextSignalCounter = 0;
			return result;
		}

		public boolean hasReservedNextSignal() {
			return reservedNextSignal != null;
		}

		public void processReservedNextSignal() {
			subscriber.onNext(reservedNextSignal);
			reservedNextSignal = null;
		}

		public boolean isTerminalSignalReceived() {
			return isTerminalSignalReceived;
		}
	}

	@Override
	public int getMode() {
		return INNER;
	}


	public SignalPoller(Context context,
						ServiceMessageSender serviceMessageSender,
						Subscriber<? super Buffer> subscriber,
						AeronInfra aeronInfra,
						Consumer<Boolean> shutdownTask) {

		this.context = context;
		this.serviceMessageSender = serviceMessageSender;
		this.subscriber = subscriber;
		this.aeronInfra = aeronInfra;
		this.shutdownTask = shutdownTask;
		this.signalFragmentHandler = new SignalFragmentHandler();
		this.demandTracker = new DemandTracker();
	}

	@Override
	public void run() {
		running = true;
		logger.debug("Signal poller started, sessionId: {}", serviceMessageSender.getSessionId());

		uk.co.real_logic.aeron.Subscription signalSub = aeronInfra.addSubscription(context.receiverChannel(),
				context.streamId());

		this.signalSub = signalSub;

		setSubscriberSubscription();

		final IdleStrategy idleStrategy = AeronUtils.newBackoffIdleStrategy();
		final DemandTracker demandTracker = this.demandTracker;
		boolean isTerminalSignalReceived = false;
		long demand = 0;
		try {
			while (running) {
				if (demand == 0) {
					demand = demandTracker.getAndReset();
				}

				int nFragmentsReceived = 0;
				int fragmentLimit = (int) Math.min(demand, context.signalPollerFragmentLimit());
				if (fragmentLimit == 0) {
					if (!signalFragmentHandler.hasReservedNextSignal()) {
						checkForTerminalSignal(signalSub);
					}
				} else {
					if (signalFragmentHandler.hasReservedNextSignal()) {
						signalFragmentHandler.processReservedNextSignal();
						fragmentLimit--;
						demand--;
					}
					if (fragmentLimit > 0) {
						nFragmentsReceived = signalSub.poll(signalFragmentHandler, fragmentLimit);
						demand -= signalFragmentHandler.getAndResetNextSignalCounter();
					}
				}
				idleStrategy.idle(nFragmentsReceived);

				if (signalFragmentHandler.isTerminalSignalReceived()) {
					isTerminalSignalReceived = true;
					running = false;
					break;
				}
			}
		} finally {
			aeronInfra.close(signalSub);

			logger.trace("about to execute shutdownTask");
			shutdownTask.accept(isTerminalSignalReceived);
		}

		logger.debug("Signal poller shutdown, sessionId: {}", serviceMessageSender.getSessionId());
	}

	private void setSubscriberSubscription() {
		//TODO: Possible timing issue due to ServiceMessagePoller termination
		try {
			if(running) {
				subscriber.onSubscribe(this);
			}
		} catch (Throwable t) {
			Exceptions.throwIfFatal(t);
			subscriber.onError(t);
		}
	}

	private void checkForTerminalSignal(Subscription nextCompleteSub) {
		signalFragmentHandler.checkForComplete = true;
		nextCompleteSub.poll(signalFragmentHandler, 1);
		signalFragmentHandler.checkForComplete = false;
	}

	public void shutdown() {
		running = false;
	}

	@Override
	public Iterator<?> downstreams() {
		String up1 = signalSub != null ? signalSub.channel()+"/"+ signalSub.streamId() :
				context.receiverChannel()+"/"+context.streamId();

		return Arrays.asList(up1, serviceMessageSender).iterator();
	}

	@Override
	public long downstreamCount() {
		return running ? 2 : 0;
	}

	@Override
	public boolean isCancelled() {
		return !running;
	}

	@Override
	public boolean isStarted() {
		return running;
	}

	@Override
	public boolean isTerminated() {
		return !running;
	}

	@Override
	public Object downstream() {
		return subscriber;
	}

	@Override
	public long requestedFromDownstream() {
		return demandTracker.current();
	}

	@Override
	public void request(long n) {
		if (running && BackpressureUtils.checkRequest(n, subscriber)) {
			try {
				serviceMessageSender.sendRequest(n);
				demandTracker.request(n);
			} catch (Exception e) {
				subscriber.onError(new ServiceMessagePublicationFailedException(ServiceMessageType.Request, e));
			}
		}
	}

	@Override
	public void cancel() {
		running = false;
	}
}

{'fraction_non_alphanumeric': 0.09395142577218583, 'fraction_numerical': 0.002578249883978755, 'mean_word_length': 4.088953030700604, 'pattern_counts': {'":': 0, '<': 21, '<?xml version=': 0, '>': 23, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4941071', 'n_tokens_mistral': 6472, 'n_tokens_neox': 5422, 'n_words': 1429}