<commit_before>/*
 *  Copyright (c) 2012-2013 Malhar, Inc.
 *  All Rights Reserved.
 */
package com.datatorrent.stram.util;

import com.datatorrent.lib.util.PubSubMessage;
import com.datatorrent.lib.util.PubSubWebSocketClient;
import com.ning.http.client.websocket.WebSocket;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeoutException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * <p>SharedPubSubWebSocketClient class.</p>
 *
 * @since 0.9.0
 */
public class SharedPubSubWebSocketClient extends PubSubWebSocketClient
{
  public static final String LAST_INDEX_TOPIC_PREFIX = PubSubMessage.INTERNAL_TOPIC_PREFIX + ".lastIndex";
  private static final Logger LOG = LoggerFactory.getLogger(SharedPubSubWebSocketClient.class);
  private Map<String, List<Handler>> topicHandlers = new HashMap<String, List<Handler>>();
  private long lastConnectTryTime;
  private long minWaitConnectionRetry = 5000;

  public interface Handler
  {
    public void onMessage(String type, String topic, Object data);

    public void onClose();

  }

  public SharedPubSubWebSocketClient(String uri, long timeoutMillis) throws URISyntaxException, IOException, ExecutionException, InterruptedException, TimeoutException
  {
    this.setUri(new URI(uri));
    lastConnectTryTime = System.currentTimeMillis();
    openConnection(timeoutMillis);
  }

  public synchronized void addHandler(String topic, Handler handler)
  {
    List<Handler> handlers;
    if (topicHandlers.containsKey(topic)) {
      handlers = topicHandlers.get(topic);<commit_msg>fixed NPE when auth is not configured<commit_after>/*
 *  Copyright (c) 2012-2013 Malhar, Inc.
 *  All Rights Reserved.
 */
package com.datatorrent.stram.util;

import com.datatorrent.lib.util.PubSubMessage;
import com.datatorrent.lib.util.PubSubWebSocketClient;
import com.ning.http.client.websocket.WebSocket;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeoutException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * <p>SharedPubSubWebSocketClient class.</p>
 *
 * @since 0.9.0
 */
public class SharedPubSubWebSocketClient extends PubSubWebSocketClient
{
  public static final String LAST_INDEX_TOPIC_PREFIX = PubSubMessage.INTERNAL_TOPIC_PREFIX + ".lastIndex";
  private static final Logger LOG = LoggerFactory.getLogger(SharedPubSubWebSocketClient.class);
  private final Map<String, List<Handler>> topicHandlers = new HashMap<String, List<Handler>>();
  private long lastConnectTryTime;
  private final long minWaitConnectionRetry = 5000;

  public interface Handler
  {
    public void onMessage(String type, String topic, Object data);

    public void onClose();

  }

  public SharedPubSubWebSocketClient(String uri, long timeoutMillis) throws URISyntaxException, IOException, ExecutionException, InterruptedException, TimeoutException
  {
    this.setUri(new URI(uri));
    lastConnectTryTime = System.currentTimeMillis();
    openConnection(timeoutMillis);
  }

  public synchronized void addHandler(String topic, Handler handler)
  {
    List<Handler> handlers;
    if (topicHandlers.containsKey(topic)) {
      handlers = topicHandlers.get(topic);
{'fraction_non_alphanumeric': 0.09476162448499117, 'fraction_numerical': 0.010005885815185403, 'mean_word_length': 5.377110694183865, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 17, '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': '6746947', 'n_tokens_mistral': 1030, 'n_tokens_neox': 989, 'n_words': 231}