<commit_before>package net.i2p.client.streaming;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
import java.util.concurrent.atomic.AtomicLong;

import net.i2p.I2PAppContext;
import net.i2p.client.I2PSession;
import net.i2p.data.DataHelper;
import net.i2p.data.Destination;
import net.i2p.util.Log;
import net.i2p.util.SimpleScheduler;
import net.i2p.util.SimpleTimer;
import net.i2p.util.SimpleTimer2;

/**
 * Maintain the state controlling a streaming connection between two 
 * destinations.
 *
 */
public class Connection {
    private I2PAppContext _context;
    private Log _log;
    private ConnectionManager _connectionManager;
    private Destination _remotePeer;
    private long _sendStreamId;
    private long _receiveStreamId;
    private long _lastSendTime;
    private AtomicLong _lastSendId;
    private boolean _resetReceived;
    private boolean _resetSent;
    private long _resetSentOn;
    private boolean _connected;
    private boolean _hardDisconnected;
    private MessageInputStream _inputStream;
    private MessageOutputStream _outputStream;
    private SchedulerChooser _chooser;
    private long _nextSendTime;
    private long _ackedPackets;
    private long _createdOn;
    private long _closeSentOn;
    private long _closeReceivedOn;
    private int _unackedPacketsReceived;
    private long _congestionWindowEnd;
    private long _highestAckedThrough;
    private boolean _isInbound;
    private boolean _updatedShareOpts;
    /** Packet ID (Long) to PacketLocal for sent but unacked packets */
    private final Map<Long, PacketLocal> _outboundPackets;
    private PacketQueue _outboundQueue;
    private ConnectionPacketHandler _handler;
    private ConnectionOptions _options;
    private ConnectionDataReceiver _receiver;
    private I2PSocketFull _socket;
    /** set to an error cause if the connection could not be established */
    private String _connectionError;
    private long _disconnectScheduledOn;
    private long _lastReceivedOn;
    private ActivityTimer _activityTimer;
    /** window size when we last saw congestion */
    private int _lastCongestionSeenAt;
    private long _lastCongestionTime;
    private long _lastCongestionHighestUnacked;
    private boolean _ackSinceCongestion;
    /** Notify this on connection (or connection failure) */
    private final Object _connectLock;
    /** how many messages have been resent and not yet ACKed? */
    private int _activeResends;
    private ConEvent _connectionEvent;
    private int _randomWait;
    
    private long _lifetimeBytesSent;
    private long _lifetimeBytesReceived;
    private long _lifetimeDupMessageSent;
    private long _lifetimeDupMessageReceived;
    
    public static final long MAX_RESEND_DELAY = 45*1000;
    public static final long MIN_RESEND_DELAY = 2*1000;

    /** wait up to 5 minutes after disconnection so we can ack/close packets */
    public static final int DISCONNECT_TIMEOUT = 5*60*1000;
    
    public static final int MAX_WINDOW_SIZE = 128;
    
    public Connection(I2PAppContext ctx, ConnectionManager manager, SchedulerChooser chooser, PacketQueue queue, ConnectionPacketHandler handler) {
        this(ctx, manager, chooser, queue, handler, null);
    }
    public Connection(I2PAppContext ctx, ConnectionManager manager, SchedulerChooser chooser, PacketQueue queue, ConnectionPacketHandler handler, ConnectionOptions opts) {
        _context = ctx;
        _connectionManager = manager;
        _chooser = chooser;
        _outboundQueue = queue;
        _handler = handler;
        _log = _context.logManager().getLog(Connection.class);
        _receiver = new ConnectionDataReceiver(_context, this);
        _inputStream = new MessageInputStream(_context);
        _outputStream = new MessageOutputStream(_context, _receiver, (opts == null ? Packet.MAX_PAYLOAD_SIZE : opts.getMaxMessageSize()));
        _outboundPackets = new TreeMap();
        _options = (opts != null ? opts : new ConnectionOptions());
        _outputStream.setWriteTimeout((int)_options.getWriteTimeout());
        _inputStream.setReadTimeout((int)_options.getReadTimeout());<commit_msg>add note<commit_after>package net.i2p.client.streaming;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
import java.util.concurrent.atomic.AtomicLong;

import net.i2p.I2PAppContext;
import net.i2p.client.I2PSession;
import net.i2p.data.DataHelper;
import net.i2p.data.Destination;
import net.i2p.util.Log;
import net.i2p.util.SimpleScheduler;
import net.i2p.util.SimpleTimer;
import net.i2p.util.SimpleTimer2;

/**
 * Maintain the state controlling a streaming connection between two 
 * destinations.
 *
 */
public class Connection {
    private I2PAppContext _context;
    private Log _log;
    private ConnectionManager _connectionManager;
    private Destination _remotePeer;
    private long _sendStreamId;
    private long _receiveStreamId;
    private long _lastSendTime;
    private AtomicLong _lastSendId;
    private boolean _resetReceived;
    private boolean _resetSent;
    private long _resetSentOn;
    private boolean _connected;
    private boolean _hardDisconnected;
    private MessageInputStream _inputStream;
    private MessageOutputStream _outputStream;
    private SchedulerChooser _chooser;
    private long _nextSendTime;
    private long _ackedPackets;
    private long _createdOn;
    private long _closeSentOn;
    private long _closeReceivedOn;
    private int _unackedPacketsReceived;
    private long _congestionWindowEnd;
    private long _highestAckedThrough;
    private boolean _isInbound;
    private boolean _updatedShareOpts;
    /** Packet ID (Long) to PacketLocal for sent but unacked packets */
    private final Map<Long, PacketLocal> _outboundPackets;
    private PacketQueue _outboundQueue;
    private ConnectionPacketHandler _handler;
    private ConnectionOptions _options;
    private ConnectionDataReceiver _receiver;
    private I2PSocketFull _socket;
    /** set to an error cause if the connection could not be established */
    private String _connectionError;
    private long _disconnectScheduledOn;
    private long _lastReceivedOn;
    private ActivityTimer _activityTimer;
    /** window size when we last saw congestion */
    private int _lastCongestionSeenAt;
    private long _lastCongestionTime;
    private long _lastCongestionHighestUnacked;
    private boolean _ackSinceCongestion;
    /** Notify this on connection (or connection failure) */
    private final Object _connectLock;
    /** how many messages have been resent and not yet ACKed? */
    private int _activeResends;
    private ConEvent _connectionEvent;
    private int _randomWait;
    
    private long _lifetimeBytesSent;
    private long _lifetimeBytesReceived;
    private long _lifetimeDupMessageSent;
    private long _lifetimeDupMessageReceived;
    
    public static final long MAX_RESEND_DELAY = 45*1000;
    public static final long MIN_RESEND_DELAY = 2*1000;

    /** wait up to 5 minutes after disconnection so we can ack/close packets */
    public static final int DISCONNECT_TIMEOUT = 5*60*1000;
    
    public static final int MAX_WINDOW_SIZE = 128;
    
    public Connection(I2PAppContext ctx, ConnectionManager manager, SchedulerChooser chooser, PacketQueue queue, ConnectionPacketHandler handler) {
        this(ctx, manager, chooser, queue, handler, null);
    }
    public Connection(I2PAppContext ctx, ConnectionManager manager, SchedulerChooser chooser, PacketQueue queue, ConnectionPacketHandler handler, ConnectionOptions opts) {
        _context = ctx;
        _connectionManager = manager;
        _chooser = chooser;
        _outboundQueue = queue;
        _handler = handler;
        _log = _context.logManager().getLog(Connection.class);
        _receiver = new ConnectionDataReceiver(_context, this);
        _inputStream = new MessageInputStream(_context);
        // FIXME pass through a passive flush delay setting as the 4th arg
        _outputStream = new MessageOutputStream(_context, _receiver, (opts == null ? Packet.MAX_PAYLOAD_SIZE : opts.getMaxMessageSize()));
        _outboundPackets = new TreeMap();
        _options = (opts != null ? opts : new ConnectionOptions());
        _outputStream.setWriteTimeout((int)_options.getWriteTimeout());
        _inputStream.setReadTimeout((int)_options.getReadTimeout());
{'fraction_non_alphanumeric': 0.0636000938746773, 'fraction_numerical': 0.009035437690682938, 'mean_word_length': 4.028318584070797, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, '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': '2014255', 'n_tokens_mistral': 2381, 'n_tokens_neox': 2292, 'n_words': 696}