<commit_before>package com.dm.awstasks.ssh;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;

import org.apache.log4j.Logger;

import com.jcraft.jsch.Channel;
import com.jcraft.jsch.Session;

public class ScpFileInputStream extends InputStream {

    private static final Logger LOG = Logger.getLogger(ScpFileInputStream.class);

    private final Session _session;
    private final Channel _execChannel;
    private final OutputStream _sshOutputStream;
    private final InputStream _sshInputStream;
    private long _availible;

    public ScpFileInputStream(Session session, String remoteFile) throws IOException {
        _session = session;
        String command = ScpDownloadCommand.constructScpInitCommand(remoteFile, false);
        _execChannel = JschCommand.openExecChannel(session, command);
        _sshOutputStream = _execChannel.getOutputStream();
        _sshInputStream = _execChannel.getInputStream();
        JschCommand.sendAckOk(_sshOutputStream);
        String serverResponse = ScpDownloadCommand.readServerResponse(_sshInputStream);
        if (serverResponse.charAt(0) != 'C') {
            throw new IllegalStateException("unexpected server response: " + serverResponse);
        }
        int start = 0;
        int end = serverResponse.indexOf(" ", start + 1);
        start = end + 1;
        end = serverResponse.indexOf(" ", start + 1);
        long filesize = Long.parseLong(serverResponse.substring(start, end));
        String filename = serverResponse.substring(end + 1);
        LOG.info("opening file: " + filename + " | " + filesize);
        JschCommand.sendAckOk(_sshOutputStream);
        _availible = filesize;
    }

    @Override
    public int available() throws IOException {
        return (int) _availible;
    }

    @Override
    public int read() throws IOException {
        if (_availible == 0) {
            return -1;
        }
        _availible--;
        return _sshInputStream.read();
    }

    @Override
    public void close() throws IOException {
        JschCommand.checkAcknowledgement(_sshInputStream);
        JschCommand.sendAckOk(_sshOutputStream);
        _execChannel.disconnect();
        _session.disconnect();
    }

}<commit_msg>throw exception on stream read when stream is closed already<commit_after>package com.dm.awstasks.ssh;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;

import org.apache.log4j.Logger;

import com.jcraft.jsch.Channel;
import com.jcraft.jsch.Session;

public class ScpFileInputStream extends InputStream {

    private static final Logger LOG = Logger.getLogger(ScpFileInputStream.class);

    private final Session _session;
    private final Channel _execChannel;
    private final OutputStream _sshOutputStream;
    private final InputStream _sshInputStream;
    private long _availible;

    public ScpFileInputStream(Session session, String remoteFile) throws IOException {
        _session = session;
        String command = ScpDownloadCommand.constructScpInitCommand(remoteFile, false);
        _execChannel = JschCommand.openExecChannel(session, command);
        _sshOutputStream = _execChannel.getOutputStream();
        _sshInputStream = _execChannel.getInputStream();
        JschCommand.sendAckOk(_sshOutputStream);
        String serverResponse = ScpDownloadCommand.readServerResponse(_sshInputStream);
        if (serverResponse.charAt(0) != 'C') {
            throw new IllegalStateException("unexpected server response: " + serverResponse);
        }
        int start = 0;
        int end = serverResponse.indexOf(" ", start + 1);
        start = end + 1;
        end = serverResponse.indexOf(" ", start + 1);
        long filesize = Long.parseLong(serverResponse.substring(start, end));
        String filename = serverResponse.substring(end + 1);
        LOG.info("opening file: " + filename + " | " + filesize);
        JschCommand.sendAckOk(_sshOutputStream);
        _availible = filesize;
    }

    @Override
    public int available() throws IOException {
        return (int) _availible;
    }

    @Override
    public int read() throws IOException {
        checkConnection();
        if (_availible == 0) {
            return -1;
        }
        _availible--;
        return _sshInputStream.read();
    }

    private void checkConnection() {
        if (!_session.isConnected()) {
            throw new IllegalStateException("stream is already closed");
        }
    }

    @Override
    public void close() throws IOException {
        try {
            JschCommand.checkAcknowledgement(_sshInputStream);
            JschCommand.sendAckOk(_sshOutputStream);
        } catch (IOException e) {
            // happens in the middle of read
        }
        _execChannel.disconnect();
        _session.disconnect();
    }

}
{'fraction_non_alphanumeric': 0.08976215098241985, 'fraction_numerical': 0.003722854188210962, 'mean_word_length': 2.954210956663941, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '5560570', 'n_tokens_mistral': 1310, 'n_tokens_neox': 1251, 'n_words': 311}