<commit_before>package com.relayrides.pushy;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.ssl.SslHandler;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;

public class ApnsClientThread<T extends ApnsPushNotification> extends Thread {
	
	private enum State {
		CONNECT,
		READY,
		RECONNECT,
		SHUTDOWN,
		EXIT
	};
	
	private final PushManager<T> pushManager;
	
	private volatile State state = null;
	
	private final Bootstrap bootstrap;
	private Channel channel = null;
	private int sequenceNumber = 0;
	
	private final SentNotificationBuffer<T> sentNotificationBuffer;
	private static final int SENT_NOTIFICATION_BUFFER_SIZE = 2048;
	
	public ApnsClientThread(final PushManager<T> pushManager) {
		super("ApnsClientThread");
		
		this.pushManager = pushManager;
		
		this.sentNotificationBuffer = new SentNotificationBuffer<T>(SENT_NOTIFICATION_BUFFER_SIZE);
		
		this.bootstrap = new Bootstrap();
		this.bootstrap.group(new NioEventLoopGroup());
		this.bootstrap.channel(NioSocketChannel.class);
		this.bootstrap.option(ChannelOption.SO_KEEPALIVE, true);
		
		final ApnsClientThread<T> clientThread = this;
		this.bootstrap.handler(new ChannelInitializer<SocketChannel>() {

			@Override
			protected void initChannel(final SocketChannel channel) throws Exception {
				final ChannelPipeline pipeline = channel.pipeline();
				
				if (pushManager.getEnvironment().isTlsRequired()) {
					pipeline.addLast("ssl", SslHandlerFactory.getSslHandler(pushManager.getKeyStore(), pushManager.getKeyStorePassword()));
				}
				
				pipeline.addLast("decoder", new ApnsErrorDecoder());
				pipeline.addLast("encoder", new PushNotificationEncoder<T>());
				pipeline.addLast("handler", new ApnsErrorHandler<T>(pushManager, clientThread));
			}
			
		});
	}
	
	@Override
	public void start() {
		this.state = State.CONNECT;
		
		super.start();
	}
	
	@Override
	public void run() {
		while (this.state != State.EXIT) {
			switch (this.state) {
				case CONNECT: {
					try {
						final ChannelFuture connectFuture = this.bootstrap.connect(this.pushManager.getEnvironment().getApnsHost(), this.pushManager.getEnvironment().getApnsPort()).sync();
						
						if (connectFuture.isSuccess()) {
							this.channel = connectFuture.channel();
							
							if (this.pushManager.getEnvironment().isTlsRequired()) {
								final Future<Channel> handshakeFuture = this.channel.pipeline().get(SslHandler.class).handshakeFuture().sync();
								
								if (handshakeFuture.isSuccess()) {
									this.state = State.READY;
								}
							} else {
								this.state = State.READY;
							}
						}
					} catch (InterruptedException e) {
						continue;
					}
					
					break;
				}
				
				case READY: {
					try {
						final SendableApnsPushNotification<T> sendableNotification =
								new SendableApnsPushNotification<T>(this.pushManager.getQueue().take(), this.sequenceNumber++);
								
						this.sentNotificationBuffer.addSentNotification(sendableNotification);
						
						// TODO Don't flush on every notification if we can avoid it
						this.channel.writeAndFlush(sendableNotification).addListener(new GenericFutureListener<ChannelFuture>() {

							public void operationComplete(final ChannelFuture future) {
								if (future.cause() != null) {
									pushManager.notifyListenersOfFailedDelivery(sendableNotification.getPushNotification(), future.cause());
								}
							}});
					} catch (InterruptedException e) {
						continue;
					}
					
					break;
				}
				
				case RECONNECT: {
					if (this.channel != null && this.channel.isOpen()) {
						this.channel.close();
					}
					
					try {
						this.channel.closeFuture().sync();
						this.state = State.CONNECT;
					} catch (InterruptedException e) {
						continue;
					}
					
					break;
				}
				
				case SHUTDOWN: {
					try {
						if (this.channel != null && this.channel.isOpen()) {
							this.channel.close().sync();
						}
						
						this.bootstrap.group().shutdownGracefully().sync();
					} catch (InterruptedException e) {
						continue;
					}
					
					this.state = State.EXIT;
					break;
				}
				
				case EXIT: {
					// Do nothing
					break;
				}
				
				default: {
					throw new IllegalArgumentException(String.format("Unexpected state: %S", this.getState()));
				}
			}
		}
	}
	
	protected void reconnect() {
		// We don't want to try to reconnect if we're already connecting or on our way out
		if (this.state == State.READY) {
			this.state = State.RECONNECT;
			this.interrupt();
		}
	}
	
	public void shutdown() {
		// Don't re-shut-down if we're already on our way out
		if (this.state != State.SHUTDOWN && this.state != State.EXIT) {
			this.state = State.SHUTDOWN;
			this.interrupt();
		}
	}
	
	protected SentNotificationBuffer<T> getSentNotificationBuffer() {
		return this.sentNotificationBuffer;
	}
}<commit_msg>Switched to a more thread-safe state management strategy and avoided a name collision with Thread.State.<commit_after>package com.relayrides.pushy;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.ssl.SslHandler;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;

public class ApnsClientThread<T extends ApnsPushNotification> extends Thread {
	
	private enum ClientState {
		CONNECT,
		READY,
		RECONNECT,
		SHUTDOWN,
		EXIT
	};
	
	private final PushManager<T> pushManager;
	
	private ClientState state = null;
	
	private final Bootstrap bootstrap;
	private Channel channel = null;
	private int sequenceNumber = 0;
	
	private final SentNotificationBuffer<T> sentNotificationBuffer;
	private static final int SENT_NOTIFICATION_BUFFER_SIZE = 2048;
	
	public ApnsClientThread(final PushManager<T> pushManager) {
		super("ApnsClientThread");
		
		this.pushManager = pushManager;
		
		this.sentNotificationBuffer = new SentNotificationBuffer<T>(SENT_NOTIFICATION_BUFFER_SIZE);
		
		this.bootstrap = new Bootstrap();
		this.bootstrap.group(new NioEventLoopGroup());
		this.bootstrap.channel(NioSocketChannel.class);
		this.bootstrap.option(ChannelOption.SO_KEEPALIVE, true);
		
		final ApnsClientThread<T> clientThread = this;
		this.bootstrap.handler(new ChannelInitializer<SocketChannel>() {

			@Override
			protected void initChannel(final SocketChannel channel) throws Exception {
				final ChannelPipeline pipeline = channel.pipeline();
				
				if (pushManager.getEnvironment().isTlsRequired()) {
					pipeline.addLast("ssl", SslHandlerFactory.getSslHandler(pushManager.getKeyStore(), pushManager.getKeyStorePassword()));
				}
				
				pipeline.addLast("decoder", new ApnsErrorDecoder());
				pipeline.addLast("encoder", new PushNotificationEncoder<T>());
				pipeline.addLast("handler", new ApnsErrorHandler<T>(pushManager, clientThread));
			}
			
		});
	}
	
	@Override
	public void start() {
		this.state = ClientState.CONNECT;
		
		super.start();
	}
	
	@Override
	public void run() {
		while (this.getClientState() != ClientState.EXIT) {
			switch (this.getClientState()) {
				case CONNECT: {
					try {
						final ChannelFuture connectFuture = this.bootstrap.connect(this.pushManager.getEnvironment().getApnsHost(), this.pushManager.getEnvironment().getApnsPort()).sync();
						
						if (connectFuture.isSuccess()) {
							this.channel = connectFuture.channel();
							
							if (this.pushManager.getEnvironment().isTlsRequired()) {
								final Future<Channel> handshakeFuture = this.channel.pipeline().get(SslHandler.class).handshakeFuture().sync();
								
								if (handshakeFuture.isSuccess()) {
									this.advanceToStateFromOriginStates(ClientState.READY, ClientState.CONNECT);
								}
							} else {
								this.advanceToStateFromOriginStates(ClientState.READY, ClientState.CONNECT);
							}
						}
					} catch (InterruptedException e) {
						continue;
					}
					
					break;
				}
				
				case READY: {
					try {
						final SendableApnsPushNotification<T> sendableNotification =
								new SendableApnsPushNotification<T>(this.pushManager.getQueue().take(), this.sequenceNumber++);
								
						this.sentNotificationBuffer.addSentNotification(sendableNotification);
						
						// TODO Don't flush on every notification if we can avoid it
						this.channel.writeAndFlush(sendableNotification).addListener(new GenericFutureListener<ChannelFuture>() {

							public void operationComplete(final ChannelFuture future) {
								if (future.cause() != null) {
									pushManager.notifyListenersOfFailedDelivery(sendableNotification.getPushNotification(), future.cause());
								}
							}});
					} catch (InterruptedException e) {
						continue;
					}
					
					break;
				}
				
				case RECONNECT: {
					if (this.channel != null && this.channel.isOpen()) {
						this.channel.close();
					}
					
					try {
						this.channel.closeFuture().sync();
						this.advanceToStateFromOriginStates(ClientState.CONNECT, ClientState.RECONNECT);
					} catch (InterruptedException e) {
						continue;
					}
					
					break;
				}
				
				case SHUTDOWN: {
					try {
						if (this.channel != null && this.channel.isOpen()) {
							this.channel.close().sync();
						}
						
						this.bootstrap.group().shutdownGracefully().sync();
					} catch (InterruptedException e) {
						continue;
					}
					
					this.advanceToStateFromOriginStates(ClientState.EXIT, ClientState.SHUTDOWN);
					
					break;
				}
				
				case EXIT: {
					// Do nothing
					break;
				}
				
				default: {
					throw new IllegalArgumentException(String.format("Unexpected state: %S", this.getState()));
				}
			}
		}
	}
	
	protected void reconnect() {
		// We don't want to try to reconnect if we're already connecting or on our way out
		this.advanceToStateFromOriginStates(ClientState.RECONNECT, ClientState.READY);
		this.interrupt();
	}
	
	public void shutdown() {
		// Don't re-shut-down if we're already on our way out
		this.advanceToStateFromOriginStates(ClientState.SHUTDOWN, ClientState.CONNECT, ClientState.READY, ClientState.RECONNECT);
		this.interrupt();
	}
	
	private ClientState getClientState() {
		synchronized (this.state) {
			return this.state;
		}
	}
	
	/**
	 * Sets the current state if and only if the current state is in one of the allowed origin states.
	 */
	private void advanceToStateFromOriginStates(final ClientState destinationState, final ClientState... allowableOriginStates) {
		synchronized (this.state) {
			for (final ClientState originState : allowableOriginStates) {
				if (this.state == originState) {
					this.state = destinationState;
					break;
				}
			}
		}
	}
	
	protected SentNotificationBuffer<T> getSentNotificationBuffer() {
		return this.sentNotificationBuffer;
	}
}
{'fraction_non_alphanumeric': 0.10879101596771364, 'fraction_numerical': 0.0008773469029654326, 'mean_word_length': 3.3892953407778204, 'pattern_counts': {'":': 0, '<': 31, '<?xml version=': 0, '>': 31, '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': '2649305', 'n_tokens_mistral': 4321, 'n_tokens_neox': 3266, 'n_words': 656}