<commit_before>package org.qfox.jestful.client.aio;

import org.qfox.jestful.client.Client;
import org.qfox.jestful.client.aio.connection.AioConnection;
import org.qfox.jestful.client.aio.connection.AioConnector;
import org.qfox.jestful.client.connection.Connector;
import org.qfox.jestful.client.exception.UnexpectedStatusException;
import org.qfox.jestful.client.gateway.Gateway;
import org.qfox.jestful.client.scheduler.Callback;
import org.qfox.jestful.commons.IOKit;
import org.qfox.jestful.commons.StringKit;
import org.qfox.jestful.commons.collection.CaseInsensitiveMap;
import org.qfox.jestful.core.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.net.ssl.SSLContext;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.net.URL;
import java.nio.channels.AsynchronousChannelGroup;
import java.nio.channels.AsynchronousSocketChannel;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Created by yangchangpei on 17/3/29.
 */
public class AioClient extends Client implements AioConnector {
    private final Logger logger = LoggerFactory.getLogger(this.getClass());

    private static AioClient defaultClient;
    private final int concurrency;
    private final ExecutorService cpu;
    private final AsynchronousChannelGroup aioChannelGroup;
    private final SSLContext sslContext;

    private AioClient(AioBuilder<?> builder) throws IOException {
        super(builder);
        this.concurrency = builder.concurrency;
        this.cpu = concurrency > 0 ? Executors.newFixedThreadPool(concurrency) : Executors.newCachedThreadPool();
        this.aioChannelGroup = AsynchronousChannelGroup.withThreadPool(cpu);
        this.sslContext = builder.sslContext;
    }

    public Object react(Action action) throws Exception {
        AioPromise promise = new AioPromise(action);
        AioEventListener listener = new JestfulAioEventListener(promise);
        action.getExtra().put(AioEventListener.class, listener);
        AsynchronousSocketChannel channel = AsynchronousSocketChannel.open(aioChannelGroup);
        PrepareCompletionHandler handler = new PrepareCompletionHandler(this, channel, action);
        cpu.execute(handler);
        return promise;
    }

    private class AioPromise extends BioPromise {
        private Set<Callback<Object>> callbacks;

        AioPromise(Action action) {
            super(action);
        }

        @Override
        public Object get() throws Exception {
            if (success == null) {
                synchronized (lock) {
                    if (success == null) lock.wait();
                    return get();
                }
            } else if (success) {
                return result;
            } else {
                throw exception;
            }
        }

        @Override
        public void get(Callback<Object> callback) {
            if (success == null) {
                synchronized (lock) {
                    if (success == null) {
                        if (callbacks == null) callbacks = new HashSet<Callback<Object>>();
                        callbacks.add(callback);
                    } else {
                        super.get(callback);
                    }
                }
            } else {
                super.get(callback);
            }
        }

        @Override
        public Client client() {
            return AioClient.this;
        }

        void fulfill() {
            synchronized (lock) {
                result = action.getResult().getBody().getValue();
                exception = action.getResult().getException();
                success = exception == null;
                lock.notifyAll();
            }
        }

    }

    protected Request newRequest(Action action) throws Exception {
        AioRequest request = aioConnect(action, gateway, this).getRequest();
        request.setRequestHeader("User-Agent", userAgent);
        action.getExtra().put(AioRequest.class, request);
        return request;
    }

    protected Response newResponse(Action action) throws Exception {<commit_msg>Client Request Retry Supports<commit_after>package org.qfox.jestful.client.aio;

import org.qfox.jestful.client.Client;
import org.qfox.jestful.client.aio.connection.AioConnection;
import org.qfox.jestful.client.aio.connection.AioConnector;
import org.qfox.jestful.client.connection.Connector;
import org.qfox.jestful.client.exception.UnexpectedStatusException;
import org.qfox.jestful.client.gateway.Gateway;
import org.qfox.jestful.client.scheduler.Callback;
import org.qfox.jestful.commons.IOKit;
import org.qfox.jestful.commons.StringKit;
import org.qfox.jestful.commons.collection.CaseInsensitiveMap;
import org.qfox.jestful.core.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.net.ssl.SSLContext;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.net.URL;
import java.nio.channels.AsynchronousChannelGroup;
import java.nio.channels.AsynchronousSocketChannel;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Created by yangchangpei on 17/3/29.
 */
public class AioClient extends Client implements AioConnector {
    private final Logger logger = LoggerFactory.getLogger(this.getClass());

    private static AioClient defaultClient;
    private final int concurrency;
    private final ExecutorService cpu;
    private final AsynchronousChannelGroup aioChannelGroup;
    private final SSLContext sslContext;

    private AioClient(AioBuilder<?> builder) throws IOException {
        super(builder);
        this.concurrency = builder.concurrency;
        this.cpu = concurrency > 0 ? Executors.newFixedThreadPool(concurrency) : Executors.newCachedThreadPool();
        this.aioChannelGroup = AsynchronousChannelGroup.withThreadPool(cpu);
        this.sslContext = builder.sslContext;
    }

    public Object react(Action action) throws Exception {
        AioPromise promise = new AioPromise(action);
        AioEventListener listener = new JestfulAioEventListener(promise);
        action.getExtra().put(AioEventListener.class, listener);
        AsynchronousSocketChannel channel = AsynchronousSocketChannel.open(aioChannelGroup);
        PrepareCompletionHandler handler = new PrepareCompletionHandler(this, channel, action);
        cpu.execute(handler);
        return promise;
    }

    private class AioPromise extends BioPromise {
        private Set<Callback<Object>> callbacks;

        AioPromise(Action action) {
            super(action);
        }

        @Override
        public Object get() throws Exception {
            if (success == null) {
                synchronized (lock) {
                    if (success == null) lock.wait();
                    return get();
                }
            } else if (success) {
                return result;
            } else {
                throw exception;
            }
        }

        @Override
        public void get(Callback<Object> callback) {
            if (success == null) {
                synchronized (lock) {
                    if (success == null) {
                        if (callbacks == null) callbacks = new HashSet<Callback<Object>>();
                        callbacks.add(callback);
                    } else {
                        super.get(callback);
                    }
                }
            } else {
                super.get(callback);
            }
        }

        @Override
        public Client client() {
            return AioClient.this;
        }

        void fulfill() {
            synchronized (lock) {
                result = action.getResult().getBody().getValue();
                exception = action.getResult().getException();
                success = exception == null;
                lock.notifyAll();
                if (callbacks != null) for (Callback<Object> callback : callbacks) super.get(callback);
            }
        }

    }

    protected Request newRequest(Action action) throws Exception {
        AioRequest request = aioConnect(action, gateway, this).getRequest();
        request.setRequestHeader("User-Agent", userAgent);
        action.getExtra().put(AioRequest.class, request);
        return request;
    }

    protected Response newResponse(Action action) throws Exception {
{'fraction_non_alphanumeric': 0.09298245614035087, 'fraction_numerical': 0.0018713450292397662, 'mean_word_length': 2.446594115276098, 'pattern_counts': {'":': 0, '<': 16, '<?xml version=': 0, '>': 18, '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': '1221545', 'n_tokens_mistral': 2285, 'n_tokens_neox': 2207, 'n_words': 481}