<commit_before>package io.joynr.messaging.httpoperation;

/*
 * #%L
 * %%
 * Copyright (C) 2011 - 2013 BMW Car IT GmbH
 * %%
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * 
 *      http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 * #L%
 */

import io.joynr.exceptions.JoynrChannelMissingException;
import io.joynr.exceptions.JoynrCommunicationException;
import io.joynr.exceptions.JoynrException;
import io.joynr.exceptions.JoynrShutdownException;
import io.joynr.messaging.MessageReceiver;
import io.joynr.messaging.MessagingSettings;
import io.joynr.messaging.datatypes.JoynrMessagingError;
import io.joynr.messaging.datatypes.JoynrMessagingErrorCode;
import io.joynr.messaging.util.Utilities;

import java.io.IOException;
import java.net.SocketException;
import java.text.SimpleDateFormat;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import joynr.JoynrMessage;

import org.apache.http.HttpEntity;
import org.apache.http.HttpResponse;
import org.apache.http.HttpStatus;
import org.apache.http.client.ClientProtocolException;
import org.apache.http.client.ResponseHandler;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.config.RequestConfig.Builder;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.util.EntityUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.fasterxml.jackson.core.JsonParseException;
import com.fasterxml.jackson.databind.JsonMappingException;
import com.fasterxml.jackson.databind.ObjectMapper;

/**
 * Callable to keep a long polling channel alive and to process incoming
 * messages.
 */

public class LongPollingCallable implements Callable<Void> {
    private static final Logger logger = LoggerFactory.getLogger(LongPollingCallable.class);
    final SimpleDateFormat format = new SimpleDateFormat("EEE, d MMM yyyy HH:mm:ss.SSS");

    public static final long LONGPOLLING_RETRY_INTERVAL_SECS = 5L;

    private CloseableHttpClient httpclient;
    private boolean shutdown = false;
    private MessageReceiver messageReceiver;
    private final ObjectMapper objectMapper;

    private String id = "";

    private Lock statusLock = new ReentrantLock();
    private Condition statusChanged = statusLock.newCondition();

    private Boolean longPollingDisabled;
    private MessagingSettings settings;
    private HttpConstants httpConstants;
    private String receiverId;
    private HttpGet httpget;
    protected int statusCode;
    private String statusText;
    private RequestConfig defaultRequestConfig;

    public LongPollingCallable(CloseableHttpClient httpclient,
                               RequestConfig defaultRequestConfig,
                               Boolean longPollingDisabled,
                               MessageReceiver messageReceiver,
                               ObjectMapper objectMapper,
                               MessagingSettings settings,
                               HttpConstants httpConstants,
                               String channelId,
                               String receiverId) {
        this.httpclient = httpclient;
        this.defaultRequestConfig = defaultRequestConfig;
        this.longPollingDisabled = longPollingDisabled;
        this.messageReceiver = messageReceiver;
        this.objectMapper = objectMapper;
        this.settings = settings;
        this.httpConstants = httpConstants;
        this.receiverId = receiverId;
    }

    @Override
    /**
     * Start long polling loop
     * throws ExecutionExeption in order to terminate channel properly
     * returns Void (null) in order to reconnect channel
     */
    public Void call() throws InterruptedException, JoynrException {
        logger.debug("LongPollingCallable OPENING CHANNEL: {} ", id);
        try {
            // Long Polling Loop
            while (true) {
                // start long poll if it has not been disabled.
                // Checking the disable switch and request creation has
                // to be an atomic process.
                // When the request creation is done it can be canceled
                // by a call to suspend.
                if (shutdown) {
                    throw new JoynrShutdownException("shutting down");
                }

                if (longPollingDisabled == true) {
                    statusLock.lockInterruptibly();
                    logger.trace("Waiting for long polling to be resumed.");
                    statusChanged.awaitUninterruptibly();

                } else {
                    longPoll();
                    Thread.sleep(10);
                }

            }
        } finally {
            // shutdown();
            logger.info("LongPollingCallable CHANNEL: " + id + " long poll loop exited");
        }
    }

    private void longPoll() {
        String responseBody = null;
        if (shutdown) {
            return;
        }

        try {

            responseBody = httpclient.execute(httpget, new ResponseHandler<String>() {

                @Override
                public String handleResponse(HttpResponse response) throws ClientProtocolException, IOException {
                    HttpEntity entity = response.getEntity();
                    String body = entity == null ? null : EntityUtils.toString(entity, "UTF-8");
                    statusCode = response.getStatusLine().getStatusCode();
                    statusText = response.getStatusLine().getReasonPhrase();
                    return body;
                }
            });
        } catch (IllegalStateException e) {
            logger.error("IllegalStateException in long poll: {} message: {}",
                         httpget.getURI().toASCIIString(),
                         e.getMessage());
            throw new JoynrShutdownException(e.getMessage());
        } catch (ClientProtocolException e) {
            logger.error("ClientProtocolException in long poll: {} message: {}",
                         httpget.getURI().toASCIIString(),
                         e.getMessage());
            delay();
            return;
        } catch (SocketException e) {
            logger.warn("SocketException in long poll: {} message: {} message: {}",
                        httpget.getURI().toASCIIString(),
                        e.getMessage());
            return;

        } catch (IOException e) {
            logger.warn("IOException in long poll: {} message: {}", httpget.getURI().toASCIIString(), e.getMessage());
            delay();
            return;
        } catch (Exception e) {
            logger.warn("Exception in long poll: {} message: {}", httpget.getURI().toASCIIString(), e.getMessage());
            delay();
            return;
        }

        switch (statusCode) {
        case HttpStatus.SC_OK:
            notifyDispatcher(responseBody);
            break;
        case HttpStatus.SC_NOT_FOUND:
            logger.error(responseBody);
            delay();<commit_msg>java: long poll delay on error<commit_after>package io.joynr.messaging.httpoperation;

/*
 * #%L
 * %%
 * Copyright (C) 2011 - 2013 BMW Car IT GmbH
 * %%
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * 
 *      http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 * #L%
 */

import io.joynr.exceptions.JoynrChannelMissingException;
import io.joynr.exceptions.JoynrCommunicationException;
import io.joynr.exceptions.JoynrException;
import io.joynr.exceptions.JoynrShutdownException;
import io.joynr.messaging.MessageReceiver;
import io.joynr.messaging.MessagingSettings;
import io.joynr.messaging.datatypes.JoynrMessagingError;
import io.joynr.messaging.datatypes.JoynrMessagingErrorCode;
import io.joynr.messaging.util.Utilities;

import java.io.IOException;
import java.net.SocketException;
import java.text.SimpleDateFormat;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import joynr.JoynrMessage;

import org.apache.http.HttpEntity;
import org.apache.http.HttpResponse;
import org.apache.http.HttpStatus;
import org.apache.http.client.ClientProtocolException;
import org.apache.http.client.ResponseHandler;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.config.RequestConfig.Builder;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.util.EntityUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.fasterxml.jackson.core.JsonParseException;
import com.fasterxml.jackson.databind.JsonMappingException;
import com.fasterxml.jackson.databind.ObjectMapper;

/**
 * Callable to keep a long polling channel alive and to process incoming
 * messages.
 */

public class LongPollingCallable implements Callable<Void> {
    private static final Logger logger = LoggerFactory.getLogger(LongPollingCallable.class);
    final SimpleDateFormat format = new SimpleDateFormat("EEE, d MMM yyyy HH:mm:ss.SSS");

    public static final long LONGPOLLING_RETRY_INTERVAL_SECS = 5L;

    private CloseableHttpClient httpclient;
    private boolean shutdown = false;
    private MessageReceiver messageReceiver;
    private final ObjectMapper objectMapper;

    private String id = "";

    private Lock statusLock = new ReentrantLock();
    private Condition statusChanged = statusLock.newCondition();

    private Boolean longPollingDisabled;
    private MessagingSettings settings;
    private HttpConstants httpConstants;
    private String receiverId;
    private HttpGet httpget;
    protected int statusCode;
    private String statusText;
    private RequestConfig defaultRequestConfig;

    public LongPollingCallable(CloseableHttpClient httpclient,
                               RequestConfig defaultRequestConfig,
                               Boolean longPollingDisabled,
                               MessageReceiver messageReceiver,
                               ObjectMapper objectMapper,
                               MessagingSettings settings,
                               HttpConstants httpConstants,
                               String channelId,
                               String receiverId) {
        this.httpclient = httpclient;
        this.defaultRequestConfig = defaultRequestConfig;
        this.longPollingDisabled = longPollingDisabled;
        this.messageReceiver = messageReceiver;
        this.objectMapper = objectMapper;
        this.settings = settings;
        this.httpConstants = httpConstants;
        this.receiverId = receiverId;
    }

    @Override
    /**
     * Start long polling loop
     * throws ExecutionExeption in order to terminate channel properly
     * returns Void (null) in order to reconnect channel
     */
    public Void call() throws InterruptedException, JoynrException {
        logger.debug("LongPollingCallable OPENING CHANNEL: {} ", id);
        try {
            // Long Polling Loop
            while (true) {
                // start long poll if it has not been disabled.
                // Checking the disable switch and request creation has
                // to be an atomic process.
                // When the request creation is done it can be canceled
                // by a call to suspend.
                if (shutdown) {
                    throw new JoynrShutdownException("shutting down");
                }

                if (longPollingDisabled == true) {
                    statusLock.lockInterruptibly();
                    logger.trace("Waiting for long polling to be resumed.");
                    statusChanged.awaitUninterruptibly();

                } else {
                    longPoll();
                    Thread.sleep(10);
                }

            }
        } finally {
            // shutdown();
            logger.info("LongPollingCallable CHANNEL: " + id + " long poll loop exited");
        }
    }

    private void longPoll() {
        String responseBody = null;
        if (shutdown) {
            return;
        }

        try {

            responseBody = httpclient.execute(httpget, new ResponseHandler<String>() {

                @Override
                public String handleResponse(HttpResponse response) throws ClientProtocolException, IOException {
                    HttpEntity entity = response.getEntity();
                    String body = entity == null ? null : EntityUtils.toString(entity, "UTF-8");
                    statusCode = response.getStatusLine().getStatusCode();
                    statusText = response.getStatusLine().getReasonPhrase();
                    logger.debug("Long poll returned: {} reason: url {}", statusCode, httpget.getURI().toASCIIString() );
                    return body;
                }
            });
        } catch (IllegalStateException e) {
            logger.error("IllegalStateException in long poll: {} message: {}",
                         httpget.getURI().toASCIIString(),
                         e.getMessage());
            throw new JoynrShutdownException(e.getMessage());
        } catch (ClientProtocolException e) {
            logger.error("ClientProtocolException in long poll: {} message: {}",
                         httpget.getURI().toASCIIString(),
                         e.getMessage());
            delay();
            return;
        } catch (SocketException e) {
            logger.warn("SocketException in long poll: {} message: {} message: {}",
                        httpget.getURI().toASCIIString(),
                        e.getMessage());
            delay();
            return;

        } catch (IOException e) {
            logger.warn("IOException in long poll: {} message: {}", httpget.getURI().toASCIIString(), e.getMessage());
            delay();
            return;
        } catch (Exception e) {
            logger.warn("Exception in long poll: {} message: {}", httpget.getURI().toASCIIString(), e.getMessage());
            delay();
            return;
        }

        switch (statusCode) {
        case HttpStatus.SC_OK:
            notifyDispatcher(responseBody);
            break;
        case HttpStatus.SC_NOT_FOUND:
            logger.error(responseBody);
            delay();
{'fraction_non_alphanumeric': 0.0853451650789508, 'fraction_numerical': 0.002348949497585802, 'mean_word_length': 2.284826403771967, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 6}, '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': '3233173', 'n_tokens_mistral': 3914, 'n_tokens_neox': 3695, 'n_words': 998}