<commit_before>package chat.rocket.android_ddp.rx;

import io.reactivex.BackpressureStrategy;
import io.reactivex.Flowable;
import io.reactivex.FlowableOnSubscribe;
import io.reactivex.exceptions.OnErrorNotImplementedException;
import io.reactivex.flowables.ConnectableFlowable;

import java.io.IOException;
import chat.rocket.android.log.RCLog;
import okhttp3.OkHttpClient;
import okhttp3.Request;
import okhttp3.Response;
import okhttp3.WebSocket;
import okhttp3.WebSocketListener;

public class RxWebSocket {
  private OkHttpClient httpClient;
  private WebSocket webSocket;

  public RxWebSocket(OkHttpClient client) {
    httpClient = client;
  }

  public ConnectableFlowable<RxWebSocketCallback.Base> connect(String url) {
    final Request request = new Request.Builder().url(url).build();

    return Flowable.create(
        (FlowableOnSubscribe<RxWebSocketCallback.Base>) emitter -> httpClient
            .newWebSocket(request, new WebSocketListener() {
              @Override
              public void onOpen(WebSocket webSocket1, Response response) {
                RxWebSocket.this.webSocket = webSocket1;
                emitter.onNext(new RxWebSocketCallback.Open(RxWebSocket.this.webSocket, response));
              }

              @Override
              public void onFailure(WebSocket webSocket1, Throwable err, Response response) {
                try {
                  emitter.onError(new RxWebSocketCallback.Failure(webSocket1, err, response));
                } catch (OnErrorNotImplementedException ex) {
                  RCLog.w(ex, "OnErrorNotImplementedException ignored");
                }
              }

              @Override
              public void onMessage(WebSocket webSocket1, String text) {
                emitter.onNext(new RxWebSocketCallback.Message(webSocket1, text));
              }

              @Override
              public void onClosed(WebSocket webSocket1, int code, String reason) {
                emitter.onNext(new RxWebSocketCallback.Close(webSocket1, code, reason));
                emitter.onComplete();
              }
            }),
        BackpressureStrategy.BUFFER
    ).publish();
  }

  public boolean sendText(String message) throws IOException {
    return webSocket.send(message);
  }

  public boolean close(int code, String reason) throws IOException {<commit_msg>WebSocket<commit_after>package chat.rocket.android_ddp.rx;

import io.reactivex.BackpressureStrategy;
import io.reactivex.Flowable;
import io.reactivex.FlowableOnSubscribe;
import io.reactivex.exceptions.OnErrorNotImplementedException;
import io.reactivex.flowables.ConnectableFlowable;

import java.io.IOException;
import chat.rocket.android.log.RCLog;
import okhttp3.OkHttpClient;
import okhttp3.Request;
import okhttp3.Response;
import okhttp3.WebSocket;
import okhttp3.WebSocketListener;

public class RxWebSocket {
  private OkHttpClient httpClient;
  private WebSocket webSocket;

  public RxWebSocket(OkHttpClient client) {
    httpClient = client;
  }

  public ConnectableFlowable<RxWebSocketCallback.Base> connect(String url) {
    final Request request = new Request.Builder().url(url).build();

    return Flowable.create(
        (FlowableOnSubscribe<RxWebSocketCallback.Base>) emitter -> httpClient
            .newWebSocket(request, new WebSocketListener() {
              @Override
              public void onOpen(WebSocket webSocket, Response response) {
                RxWebSocket.this.webSocket = webSocket;
                emitter.onNext(new RxWebSocketCallback.Open(RxWebSocket.this.webSocket, response));
              }

              @Override
              public void onFailure(WebSocket webSocket, Throwable err, Response response) {
                try {
                  emitter.onError(new RxWebSocketCallback.Failure(webSocket, err, response));
                } catch (OnErrorNotImplementedException ex) {
                  RCLog.w(ex, "OnErrorNotImplementedException ignored");
                }
              }

              @Override
              public void onMessage(WebSocket webSocket, String text) {
                emitter.onNext(new RxWebSocketCallback.Message(webSocket, text));
              }

              @Override
              public void onClosed(WebSocket webSocket, int code, String reason) {
                emitter.onNext(new RxWebSocketCallback.Close(webSocket, code, reason));
                emitter.onComplete();
              }
            }),
        BackpressureStrategy.BUFFER
    ).publish();
  }

  public boolean sendText(String message) throws IOException {
    return webSocket.send(message);
  }

  public boolean close(int code, String reason) throws IOException {
{'fraction_non_alphanumeric': 0.08017057569296375, 'fraction_numerical': 0.003837953091684435, 'mean_word_length': 2.731901352426412, 'pattern_counts': {'":': 0, '<': 7, '<?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': '419113', 'n_tokens_mistral': 1274, 'n_tokens_neox': 1195, 'n_words': 262}