<commit_before>package com.miguelgaeta.bootstrap.mg_websocket;

import com.miguelgaeta.bootstrap.mg_delay.MGDelay;
import com.miguelgaeta.bootstrap.mg_log.MGLog;
import com.miguelgaeta.bootstrap.mg_preference.MGPreferenceRx;
import com.miguelgaeta.bootstrap.mg_rest.MGRestClient;
import com.miguelgaeta.bootstrap.mg_rx.MGRxError;
import com.miguelgaeta.bootstrap.mg_websocket.events.MGWebsocketEventClosed;
import com.miguelgaeta.bootstrap.mg_websocket.events.MGWebsocketEventError;
import com.miguelgaeta.bootstrap.mg_websocket.events.MGWebsocketEventMessage;
import com.miguelgaeta.bootstrap.mg_websocket.events.MGWebsocketEventOpened;

import org.java_websocket.WebSocket;
import org.java_websocket.client.WebSocketClient;
import org.java_websocket.handshake.ServerHandshake;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import javax.net.ssl.SSLSocketFactory;

import lombok.AccessLevel;
import lombok.Getter;
import lombok.NonNull;
import rx.Observable;
import rx.Subscription;

/**
 * Created by mrkcsc on 4/15/15.
 */
class MGWebsocketClient {

    @Getter(lazy = true) private final MGPreferenceRx<MGWebsocketEventOpened>  eventOpn = MGPreferenceRx.create(null);
    @Getter(lazy = true) private final MGPreferenceRx<MGWebsocketEventClosed>  eventCls = MGPreferenceRx.create(null);
    @Getter(lazy = true) private final MGPreferenceRx<MGWebsocketEventMessage> eventMsg = MGPreferenceRx.create(null);
    @Getter(lazy = true) private final MGPreferenceRx<MGWebsocketEventError>   eventErr = MGPreferenceRx.create(null);

    @Getter(AccessLevel.PACKAGE)
    private final List<String> messageBuffer = new ArrayList<>();

    // Raw websocket object.
    private WebSocketClient client;

    // Optional reconnect delay.
    private Integer reconnectDelay;

    // Connect to url.
    private String url;

    // Subscription to keep alive observable.
    private Subscription heartBeatSubscription;

    // Optionally used for SSL.
    private SSLSocketFactory socketFactory;

    /**
     * This is the state we want the client to be in. If
     * reconnecting, will always try to achieve this state.
     */
    private MGWebsocketState clientDesiredState = MGWebsocketState.NOT_YET_CONNECTED;

    /**
     * Default constructor.
     */
    public MGWebsocketClient() {

        // Reconnection logic is powered by observable
        // events to only register once per instance.
        configureReconnect();
    }

    /**
     * Connect to the web socket given a
     * specified url.  Optionally provide a
     * specified reconnection delay.=
     */
    public void connect(@NonNull String url, Integer reconnectDelay, SSLSocketFactory socketFactory) {

        this.url = url;
        this.reconnectDelay = reconnectDelay;
        this.socketFactory = socketFactory;

        client = createClient();
        client.connect();
        clientDesiredState = MGWebsocketState.OPENED;
    }

    /**
     * Close the socket.
     */
    public void disconnect() {

        // Don't need to call close if already happened or happening.
        boolean closed =
            getState() == MGWebsocketState.CLOSED  ||
            getState() == MGWebsocketState.CLOSING ||
            getState() == MGWebsocketState.NOT_YET_CONNECTED;

        if (client != null && !closed) {
            client.close();
            clientDesiredState = MGWebsocketState.CLOSED;
        }
    }

    /**
     * Send a string to the web socket.  If
     * buffer flag is set, send the message as
     * soon as web socket is connected.
     */
    public void message(@NonNull String message, boolean buffered) {

        if (client != null && client.getReadyState() == WebSocket.READYSTATE.OPEN) {
            client.send(message);

        } else if (buffered) {

            // Send when connected.
            messageBuffer.add(message);
        }
    }

    /**
     * Send an arbitrary object as a json string.
     */
    public void message(@NonNull Object message, boolean buffered) {

        Observable.create(subscriber -> {

            // Send serialized message.
            message(MGRestClient.getGson().toJson(message), buffered);

            // Done with observable.
            subscriber.onCompleted();

        }).subscribe(r -> { }, MGRxError.create(null, "Unable to send message."));
    }

    /**
     * Provides new heartbeat parameters.
     */
    public void heartBeat(Integer heartBeatInterval, String message) {

        // If the socket is opened, try to keep it
        // open by sending keep alive messages on
        // an interval defined by the user.
        configureHeartbeat(heartBeatInterval, message);
    }

    /**
     * Returns the current state of the
     * web socket client.
     */
    public MGWebsocketState getState() {

        if (client == null) {

            return MGWebsocketState.NOT_YET_CONNECTED;
        }

        switch (client.getReadyState()) {

            case CONNECTING:
                return MGWebsocketState.CONNECTING;

            case OPEN:
                return MGWebsocketState.OPENED;

            case CLOSING:
                return MGWebsocketState.CLOSING;

            case CLOSED:
                return MGWebsocketState.CLOSED;

            case NOT_YET_CONNECTED:
            default:
                return MGWebsocketState.NOT_YET_CONNECTED;
        }
    }

    /**
     * Configures the native websocket
     * client implementation used by
     * the wrapper.
     */
    private WebSocketClient createClient() {

        WebSocketClient client = new MGWebsocketClientBase(url) {

            @Override
            public void onDidOpen(ServerHandshake handshakeData) {

                getEventCls().set(null);
                getEventErr().set(null);
                getEventOpn().set(MGWebsocketEventOpened.create(handshakeData.getHttpStatus(), handshakeData.getHttpStatusMessage()));
            }

            @Override
            public void onDidMessage(String message) {

                getEventMsg().set(MGWebsocketEventMessage.create(message));
            }

            @Override
            public void onDidClose(int code, String reason, boolean remote) {

                getEventOpn().set(null);
                getEventErr().set(null);
                getEventCls().set(MGWebsocketEventClosed.create(code, reason, remote));
            }

            @Override
            public void onDidError(Exception exception) {<commit_msg>Fixed a bug where web socket could send a message when not yet connected.  Added a guard against the exception in the rare case it does happen.<commit_after>package com.miguelgaeta.bootstrap.mg_websocket;

import com.miguelgaeta.bootstrap.mg_delay.MGDelay;
import com.miguelgaeta.bootstrap.mg_log.MGLog;
import com.miguelgaeta.bootstrap.mg_preference.MGPreferenceRx;
import com.miguelgaeta.bootstrap.mg_rest.MGRestClient;
import com.miguelgaeta.bootstrap.mg_rx.MGRxError;
import com.miguelgaeta.bootstrap.mg_websocket.events.MGWebsocketEventClosed;
import com.miguelgaeta.bootstrap.mg_websocket.events.MGWebsocketEventError;
import com.miguelgaeta.bootstrap.mg_websocket.events.MGWebsocketEventMessage;
import com.miguelgaeta.bootstrap.mg_websocket.events.MGWebsocketEventOpened;

import org.java_websocket.WebSocket;
import org.java_websocket.handshake.ServerHandshake;

import java.io.IOException;
import java.nio.channels.NotYetConnectedException;
import java.util.ArrayList;
import java.util.List;

import javax.net.ssl.SSLSocketFactory;

import lombok.AccessLevel;
import lombok.Getter;
import lombok.NonNull;
import rx.Observable;
import rx.Subscription;

/**
 * Created by mrkcsc on 4/15/15.
 */
class MGWebsocketClient {

    @Getter(lazy = true) private final MGPreferenceRx<MGWebsocketEventOpened>  eventOpn = MGPreferenceRx.create(null);
    @Getter(lazy = true) private final MGPreferenceRx<MGWebsocketEventClosed>  eventCls = MGPreferenceRx.create(null);
    @Getter(lazy = true) private final MGPreferenceRx<MGWebsocketEventMessage> eventMsg = MGPreferenceRx.create(null);
    @Getter(lazy = true) private final MGPreferenceRx<MGWebsocketEventError>   eventErr = MGPreferenceRx.create(null);

    @Getter(AccessLevel.PACKAGE)
    private final List<String> messageBuffer = new ArrayList<>();

    // Raw websocket object.
    private MGWebsocketClientBase client;

    // Optional reconnect delay.
    private Integer reconnectDelay;

    // Connect to url.
    private String url;

    // Subscription to keep alive observable.
    private Subscription heartBeatSubscription;

    // Optionally used for SSL.
    private SSLSocketFactory socketFactory;

    /**
     * This is the state we want the client to be in. If
     * reconnecting, will always try to achieve this state.
     */
    private MGWebsocketState clientDesiredState = MGWebsocketState.NOT_YET_CONNECTED;

    /**
     * Default constructor.
     */
    public MGWebsocketClient() {

        // Reconnection logic is powered by observable
        // events to only register once per instance.
        configureReconnect();
    }

    /**
     * Connect to the web socket given a
     * specified url.  Optionally provide a
     * specified reconnection delay.=
     */
    public void connect(@NonNull String url, Integer reconnectDelay, SSLSocketFactory socketFactory) {

        this.url = url;
        this.reconnectDelay = reconnectDelay;
        this.socketFactory = socketFactory;

        client = createClient();
        client.connect();
        clientDesiredState = MGWebsocketState.OPENED;
    }

    /**
     * Close the socket.
     */
    public void disconnect() {

        // Don't need to call close if already happened or happening.
        boolean closed =
            getState() == MGWebsocketState.CLOSED  ||
            getState() == MGWebsocketState.CLOSING ||
            getState() == MGWebsocketState.NOT_YET_CONNECTED;

        if (client != null && !closed) {
            client.close();
            clientDesiredState = MGWebsocketState.CLOSED;
        }
    }

    /**
     * Send a string to the web socket.  If
     * buffer flag is set, send the message as
     * soon as web socket is connected.
     */
    public void message(@NonNull String message, boolean buffered) {

        try {

            if (client != null && client.getReadyState() == WebSocket.READYSTATE.OPEN) {
                client.send(message);

            } else if (buffered) {

                // Send when connected.
                messageBuffer.add(message);
            }

        } catch (NotYetConnectedException e) {

            MGLog.e("Tried to send message when not connected.", e);
        }
    }

    /**
     * Send an arbitrary object as a json string.
     */
    public void message(@NonNull Object message, boolean buffered) {

        Observable.create(subscriber -> {

            // Send serialized message.
            message(MGRestClient.getGson().toJson(message), buffered);

            // Done with observable.
            subscriber.onCompleted();

        }).subscribe(r -> { }, MGRxError.create(null, "Unable to send message."));
    }

    /**
     * Provides new heartbeat parameters.
     */
    public void heartBeat(Integer heartBeatInterval, String message) {

        // If the socket is opened, try to keep it
        // open by sending keep alive messages on
        // an interval defined by the user.
        configureHeartbeat(heartBeatInterval, message);
    }

    /**
     * Returns the current state of the
     * web socket client.
     */
    public MGWebsocketState getState() {

        if (client == null) {

            return MGWebsocketState.NOT_YET_CONNECTED;
        }

        switch (client.getReadyState()) {

            case CONNECTING:
                return MGWebsocketState.CONNECTING;

            case OPEN:
                return MGWebsocketState.OPENED;

            case CLOSING:
                return MGWebsocketState.CLOSING;

            case CLOSED:
                return MGWebsocketState.CLOSED;

            case NOT_YET_CONNECTED:
            default:
                return MGWebsocketState.NOT_YET_CONNECTED;
        }
    }

    /**
     * Configures the native websocket
     * client implementation used by
     * the wrapper.
     */
    private MGWebsocketClientBase createClient() {

        MGWebsocketClientBase client = new MGWebsocketClientBase(url) {

            @Override
            public void onDidOpen(ServerHandshake handshakeData) {

                getEventCls().set(null);
                getEventErr().set(null);
                getEventOpn().set(MGWebsocketEventOpened.create(handshakeData.getHttpStatus(), handshakeData.getHttpStatusMessage()));
            }

            @Override
            public void onDidMessage(String message) {

                getEventMsg().set(MGWebsocketEventMessage.create(message));
            }

            @Override
            public void onDidClose(int code, String reason, boolean remote) {

                getEventOpn().set(null);
                getEventErr().set(null);
                getEventCls().set(MGWebsocketEventClosed.create(code, reason, remote));
            }

            @Override
            public void onDidError(Exception exception) {
{'fraction_non_alphanumeric': 0.086373228821224, 'fraction_numerical': 0.0007536930961712391, 'mean_word_length': 2.474469756480754, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 19, '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': '987832', 'n_tokens_mistral': 3780, 'n_tokens_neox': 3647, 'n_words': 888}