<commit_before>package org.jboss.cx.remoting.http.urlconnection;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.HttpURLConnection;
import java.net.URL;
import java.util.List;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.FutureTask;
import java.util.concurrent.Future;
import org.jboss.cx.remoting.http.AbstractHttpChannel;
import org.jboss.cx.remoting.http.cookie.CookieClientSession;
import org.jboss.cx.remoting.http.HttpMessageWriter;
import org.jboss.cx.remoting.http.RemotingHttpChannelContext;
import org.jboss.cx.remoting.log.Logger;
import org.jboss.cx.remoting.util.AbstractOutputStreamByteMessageOutput;
import org.jboss.cx.remoting.util.IoUtil;
import org.jboss.cx.remoting.util.NamingThreadFactory;

/**
 *
 */
public abstract class AbstractHttpUrlChannel extends AbstractHttpChannel {

    private static final Logger log = Logger.getLogger(AbstractHttpUrlChannel.class);

    private final CookieClientSession cookieClientSession = new CookieClientSession();

    protected AbstractHttpUrlChannel() {
    }

    // Configuration

    private int concurrentRequests = 2;
    private int connectTimeout = 5000;
    private int readTimeout = 0;  // Default to unlimited to support "parking" the connection at the other end
    private int errorBackoffTime = 5000;
    private URL connectUrl;

    public int getConcurrentRequests() {
        return concurrentRequests;
    }

    public void setConcurrentRequests(final int concurrentRequests) {
        this.concurrentRequests = concurrentRequests;
    }

    public int getConnectTimeout() {
        return connectTimeout;
    }

    public void setConnectTimeout(final int connectTimeout) {
        this.connectTimeout = connectTimeout;
    }

    public int getReadTimeout() {
        return readTimeout;
    }

    public void setReadTimeout(final int readTimeout) {
        this.readTimeout = readTimeout;
    }

    public int getErrorBackoffTime() {
        return errorBackoffTime;
    }

    public void setErrorBackoffTime(final int errorBackoffTime) {
        this.errorBackoffTime = errorBackoffTime;
    }

    public URL getConnectUrl() {
        return connectUrl;
    }

    public void setConnectUrl(final URL connectUrl) {
        this.connectUrl = connectUrl;
    }

    // Dependencies

    private Executor executor;

    public Executor getExecutor() {
        return executor;
    }

    public void setExecutor(final Executor executor) {
        this.executor = executor;
    }

    // Lifecycle

    private ExecutorService executorService;
    private Future[] futures;

    public void create() {
        super.create();
        if (executor == null) {
            executor = executorService = Executors.newFixedThreadPool(concurrentRequests, new NamingThreadFactory(Executors.defaultThreadFactory(), "Remoting HTTP client %s"));
        }
        if (connectUrl == null) {
            throw new NullPointerException("connectUrl is null");
        }
    }

    public void start() {
        final Future[] futures = new Future[concurrentRequests];
        for (int i = 0; i < futures.length; i++) {
            final FutureTask task = new FutureTask<Void>(null) {
                public void run() {
                    while (! isCancelled()) try {
                        handleRequest();
                    } catch (Throwable t) {
                        log.trace(t, "Request hander failed");
                    }
                }
            };
            executor.execute(task);
            futures[i] = task;
        }
        this.futures = futures;
    }

    public void stop() {
        if (futures != null) {
            final Future[] futures = this.futures;
            this.futures = null;
            for (Future future : futures) try {
                future.cancel(true);
            } catch (Throwable t) {
                log.trace(t, "Error cancelling task");
            }
        }
    }

    public void destroy() {
        try {
            super.destroy();
        } finally {
            if (executorService != null) {
                executorService.shutdown();
            }
        }
        executor = executorService = null;
    }

    // Interface

    protected void handleRequest() {
        final URL connectUrl = getConnectUrl();
        final RemotingHttpChannelContext channelContext = getChannelContext();
        final int localParkTime = getLocalParkTime();
        final int remoteParkTime = getRemoteParkTime();
        final HttpMessageWriter messageWriter = channelContext.waitForOutgoingHttpMessage(localParkTime);
        try {
            final HttpURLConnection httpConnection = intializeConnection(connectUrl);
            try {
                if (remoteParkTime >= 0) {
                    httpConnection.addRequestProperty("Park-Timeout", Integer.toString(remoteParkTime));
                }
                httpConnection.connect();
                final OutputStream outputStream = httpConnection.getOutputStream();
                try {
                    messageWriter.writeMessageData(new AbstractOutputStreamByteMessageOutput(outputStream) {
                        public void commit() throws IOException {
                        }
                    });
                    // now read the reply
                    final String responseMessage = httpConnection.getResponseMessage();
                    log.trace("HTTP server sent back a response message: %s", responseMessage);
                    final List<String> setCookies = httpConnection.getHeaderFields().get("Set-Cookie");
                    for (String s : setCookies) {
                        cookieClientSession.handleSetCookieHeader(s, connectUrl.getHost(), connectUrl.getPath());
                    }
                    final InputStream inputStream = httpConnection.getInputStream();
                    try {
                        channelContext.processInboundMessage(INIT_ME);
                    } finally {
                        IoUtil.closeSafely(inputStream);
                    }
                } finally {
                    IoUtil.closeSafely(outputStream);
                }
            } catch (IOException e) {
                // probably a HTTP error occurred, so let's consume it
                try {
                    final InputStream errorStream = httpConnection.getErrorStream();
                    if (errorStream != null) try {<commit_msg>Fix an unfinished refactor<commit_after>package org.jboss.cx.remoting.http.urlconnection;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.HttpURLConnection;
import java.net.URL;
import java.util.List;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.FutureTask;
import java.util.concurrent.Future;
import org.jboss.cx.remoting.http.AbstractHttpChannel;
import org.jboss.cx.remoting.http.cookie.CookieClientSession;
import org.jboss.cx.remoting.http.HttpMessageWriter;
import org.jboss.cx.remoting.http.RemotingHttpChannelContext;
import org.jboss.cx.remoting.log.Logger;
import org.jboss.cx.remoting.util.AbstractOutputStreamByteMessageOutput;
import org.jboss.cx.remoting.util.IoUtil;
import org.jboss.cx.remoting.util.NamingThreadFactory;
import org.jboss.cx.remoting.util.InputStreamByteMessageInput;

/**
 *
 */
public abstract class AbstractHttpUrlChannel extends AbstractHttpChannel {

    private static final Logger log = Logger.getLogger(AbstractHttpUrlChannel.class);

    private final CookieClientSession cookieClientSession = new CookieClientSession();

    protected AbstractHttpUrlChannel() {
    }

    // Configuration

    private int concurrentRequests = 2;
    private int connectTimeout = 5000;
    private int readTimeout = 0;  // Default to unlimited to support "parking" the connection at the other end
    private int errorBackoffTime = 5000;
    private URL connectUrl;

    public int getConcurrentRequests() {
        return concurrentRequests;
    }

    public void setConcurrentRequests(final int concurrentRequests) {
        this.concurrentRequests = concurrentRequests;
    }

    public int getConnectTimeout() {
        return connectTimeout;
    }

    public void setConnectTimeout(final int connectTimeout) {
        this.connectTimeout = connectTimeout;
    }

    public int getReadTimeout() {
        return readTimeout;
    }

    public void setReadTimeout(final int readTimeout) {
        this.readTimeout = readTimeout;
    }

    public int getErrorBackoffTime() {
        return errorBackoffTime;
    }

    public void setErrorBackoffTime(final int errorBackoffTime) {
        this.errorBackoffTime = errorBackoffTime;
    }

    public URL getConnectUrl() {
        return connectUrl;
    }

    public void setConnectUrl(final URL connectUrl) {
        this.connectUrl = connectUrl;
    }

    // Dependencies

    private Executor executor;

    public Executor getExecutor() {
        return executor;
    }

    public void setExecutor(final Executor executor) {
        this.executor = executor;
    }

    // Lifecycle

    private ExecutorService executorService;
    private Future[] futures;

    public void create() {
        super.create();
        if (executor == null) {
            executor = executorService = Executors.newFixedThreadPool(concurrentRequests, new NamingThreadFactory(Executors.defaultThreadFactory(), "Remoting HTTP client %s"));
        }
        if (connectUrl == null) {
            throw new NullPointerException("connectUrl is null");
        }
    }

    public void start() {
        final Future[] futures = new Future[concurrentRequests];
        for (int i = 0; i < futures.length; i++) {
            final FutureTask task = new FutureTask<Void>(null) {
                public void run() {
                    while (! isCancelled()) try {
                        handleRequest();
                    } catch (Throwable t) {
                        log.trace(t, "Request hander failed");
                    }
                }
            };
            executor.execute(task);
            futures[i] = task;
        }
        this.futures = futures;
    }

    public void stop() {
        if (futures != null) {
            final Future[] futures = this.futures;
            this.futures = null;
            for (Future future : futures) try {
                future.cancel(true);
            } catch (Throwable t) {
                log.trace(t, "Error cancelling task");
            }
        }
    }

    public void destroy() {
        try {
            super.destroy();
        } finally {
            if (executorService != null) {
                executorService.shutdown();
            }
        }
        executor = executorService = null;
    }

    // Interface

    protected void handleRequest() {
        final URL connectUrl = getConnectUrl();
        final RemotingHttpChannelContext channelContext = getChannelContext();
        final int localParkTime = getLocalParkTime();
        final int remoteParkTime = getRemoteParkTime();
        final HttpMessageWriter messageWriter = channelContext.waitForOutgoingHttpMessage(localParkTime);
        try {
            final HttpURLConnection httpConnection = intializeConnection(connectUrl);
            try {
                if (remoteParkTime >= 0) {
                    httpConnection.addRequestProperty("Park-Timeout", Integer.toString(remoteParkTime));
                }
                httpConnection.connect();
                final OutputStream outputStream = httpConnection.getOutputStream();
                try {
                    messageWriter.writeMessageData(new AbstractOutputStreamByteMessageOutput(outputStream) {
                        public void commit() throws IOException {
                        }
                    });
                    // now read the reply
                    final String responseMessage = httpConnection.getResponseMessage();
                    log.trace("HTTP server sent back a response message: %s", responseMessage);
                    final List<String> setCookies = httpConnection.getHeaderFields().get("Set-Cookie");
                    for (String s : setCookies) {
                        cookieClientSession.handleSetCookieHeader(s, connectUrl.getHost(), connectUrl.getPath());
                    }
                    final InputStream inputStream = httpConnection.getInputStream();
                    try {
                        channelContext.processInboundMessage(new InputStreamByteMessageInput(inputStream, -1));
                    } finally {
                        IoUtil.closeSafely(inputStream);
                    }
                } finally {
                    IoUtil.closeSafely(outputStream);
                }
            } catch (IOException e) {
                // probably a HTTP error occurred, so let's consume it
                try {
                    final InputStream errorStream = httpConnection.getErrorStream();
                    if (errorStream != null) try {
{'fraction_non_alphanumeric': 0.08265961465810351, 'fraction_numerical': 0.0018889308651303363, 'mean_word_length': 2.1143529411764708, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 9, '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': '6966544', 'n_tokens_mistral': 3393, 'n_tokens_neox': 3182, 'n_words': 773}