<commit_before>package org.spacehq.packetlib.tcp;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.handler.timeout.ReadTimeoutException;
import io.netty.handler.timeout.ReadTimeoutHandler;
import io.netty.handler.timeout.TimeoutException;
import io.netty.handler.timeout.WriteTimeoutHandler;
import org.spacehq.packetlib.*;
import org.spacehq.packetlib.event.session.*;
import org.spacehq.packetlib.packet.Packet;
import org.spacehq.packetlib.packet.PacketProtocol;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;

public class TcpSession extends SimpleChannelInboundHandler<Packet> implements Session {

	private String host;
	private int port;
	private PacketProtocol protocol;
	private Bootstrap bootstrap;
	private EventLoopGroup group;
	private Channel channel;
	private boolean disconnected = false;
	private boolean writing = false;
	private int compressionThreshold = -1;
	private int readTimeout = 30;
	private int writeTimeout = 0;
	private TimeoutHandler timeoutHandler = null;
	private ConnectTimeoutHandlerContainer container;
	private List<Packet> packets = new ArrayList<Packet>();

	private Map<String, Object> flags = new HashMap<String, Object>();
	private List<SessionListener> listeners = new ArrayList<SessionListener>();

	public TcpSession(String host, int port, PacketProtocol protocol, EventLoopGroup group, Bootstrap bootstrap, ConnectTimeoutHandlerContainer container) {
		this.host = host;
		this.port = port;
		this.protocol = protocol;
		this.group = group;
		this.bootstrap = bootstrap;
		this.container = container;
	}

	@Override
	public void connect() {
		this.connect(true);
	}

	@Override
	public void connect(boolean wait) {
		if(this.bootstrap == null) {
			if(!this.disconnected) {
				return;
			} else {
				throw new IllegalStateException("Session has already been disconnected.");
			}
		}

		final AtomicBoolean calledTimeout = new AtomicBoolean(false);
		try {
			this.bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, this.container.getConnectTimeout());
			ChannelFuture future = this.bootstrap.connect();
			this.bootstrap = null;
			if(wait) {
				while(this.channel == null && !this.disconnected) {
					try {
						Thread.sleep(5);
					} catch(InterruptedException e) {
					}
				}
			} else {
				future.addListener(new ChannelFutureListener() {
					@Override
					public void operationComplete(ChannelFuture channelFuture) throws Exception {
						if(channelFuture.cause() instanceof ConnectTimeoutException && container.getConnectTimeoutHandler() != null && !calledTimeout.get()) {
							container.getConnectTimeoutHandler().onTimeout(TcpSession.this, TimeoutType.CONNECT);
							calledTimeout.set(true);
						}
					}
				});
			}
		} catch(Exception e) {
			if(e instanceof ConnectTimeoutException && container.getConnectTimeoutHandler() != null && !calledTimeout.get()) {
				container.getConnectTimeoutHandler().onTimeout(TcpSession.this, TimeoutType.CONNECT);
				calledTimeout.set(true);
			} else {
				System.err.println("Failed to establish connection.");
				e.printStackTrace();
			}
		}
	}

	@Override
	public String getHost() {
		return this.host;
	}

	@Override
	public int getPort() {
		return this.port;
	}

	@Override
	public PacketProtocol getPacketProtocol() {<commit_msg>Timeout should be multiplied by 1000 for milliseconds.<commit_after>package org.spacehq.packetlib.tcp;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.handler.timeout.ReadTimeoutException;
import io.netty.handler.timeout.ReadTimeoutHandler;
import io.netty.handler.timeout.TimeoutException;
import io.netty.handler.timeout.WriteTimeoutHandler;
import org.spacehq.packetlib.*;
import org.spacehq.packetlib.event.session.*;
import org.spacehq.packetlib.packet.Packet;
import org.spacehq.packetlib.packet.PacketProtocol;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;

public class TcpSession extends SimpleChannelInboundHandler<Packet> implements Session {

	private String host;
	private int port;
	private PacketProtocol protocol;
	private Bootstrap bootstrap;
	private EventLoopGroup group;
	private Channel channel;
	private boolean disconnected = false;
	private boolean writing = false;
	private int compressionThreshold = -1;
	private int readTimeout = 30;
	private int writeTimeout = 0;
	private TimeoutHandler timeoutHandler = null;
	private ConnectTimeoutHandlerContainer container;
	private List<Packet> packets = new ArrayList<Packet>();

	private Map<String, Object> flags = new HashMap<String, Object>();
	private List<SessionListener> listeners = new ArrayList<SessionListener>();

	public TcpSession(String host, int port, PacketProtocol protocol, EventLoopGroup group, Bootstrap bootstrap, ConnectTimeoutHandlerContainer container) {
		this.host = host;
		this.port = port;
		this.protocol = protocol;
		this.group = group;
		this.bootstrap = bootstrap;
		this.container = container;
	}

	@Override
	public void connect() {
		this.connect(true);
	}

	@Override
	public void connect(boolean wait) {
		if(this.bootstrap == null) {
			if(!this.disconnected) {
				return;
			} else {
				throw new IllegalStateException("Session has already been disconnected.");
			}
		}

		final AtomicBoolean calledTimeout = new AtomicBoolean(false);
		try {
			this.bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, this.container.getConnectTimeout() * 1000);
			ChannelFuture future = this.bootstrap.connect();
			this.bootstrap = null;
			if(wait) {
				while(this.channel == null && !this.disconnected) {
					try {
						Thread.sleep(5);
					} catch(InterruptedException e) {
					}
				}
			} else {
				future.addListener(new ChannelFutureListener() {
					@Override
					public void operationComplete(ChannelFuture channelFuture) throws Exception {
						if(channelFuture.cause() instanceof ConnectTimeoutException && container.getConnectTimeoutHandler() != null && !calledTimeout.get()) {
							calledTimeout.set(true);
							container.getConnectTimeoutHandler().onTimeout(TcpSession.this, TimeoutType.CONNECT);
						}
					}
				});
			}
		} catch(Exception e) {
			if(e instanceof ConnectTimeoutException && container.getConnectTimeoutHandler() != null && !calledTimeout.get()) {
				calledTimeout.set(true);
				container.getConnectTimeoutHandler().onTimeout(TcpSession.this, TimeoutType.CONNECT);
			} else {
				System.err.println("Failed to establish connection.");
				e.printStackTrace();
			}
		}
	}

	@Override
	public String getHost() {
		return this.host;
	}

	@Override
	public int getPort() {
		return this.port;
	}

	@Override
	public PacketProtocol getPacketProtocol() {
{'fraction_non_alphanumeric': 0.10398359695371998, 'fraction_numerical': 0.0026362038664323375, 'mean_word_length': 4.657829328914665, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 17, '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': '5040884', 'n_tokens_mistral': 2181, 'n_tokens_neox': 1892, 'n_words': 450}