<commit_before>/*
 * Copyright 2017 <NAME> <<EMAIL>>
 *
 * 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.
 */

package com.ivkos.gpsd4j.client;

import com.ivkos.gpsd4j.messages.GpsdCommandMessage;
import com.ivkos.gpsd4j.messages.GpsdMessage;
import com.ivkos.gpsd4j.messages.WatchMessage;
import com.ivkos.gpsd4j.support.GpsdParseException;
import io.vertx.core.AsyncResult;
import io.vertx.core.Vertx;
import io.vertx.core.net.NetClient;
import io.vertx.core.net.NetClientOptions;
import io.vertx.core.net.NetSocket;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.locks.StampedLock;
import java.util.function.Consumer;

import static com.ivkos.gpsd4j.support.SerializationHelper.*;
import static java.lang.String.format;
import static java.util.Collections.*;
import static java.util.Objects.requireNonNull;

public class GpsdClient
{
   private static final Logger log = LoggerFactory.getLogger(GpsdClient.class);

   private static final int BUFFER_SIZE = 4 * 1024;

   private final Map<Class<? extends GpsdMessage>, List<Consumer<GpsdMessage>>> handlers =
         synchronizedMap(new HashMap<>());

   private final StampedLock startingLock = new StampedLock();
   private final StampedLock stoppingLock = new StampedLock();

   private final String serverHost;
   private final int serverPort;
   private final GpsdClientOptions options;

   private boolean running = false;
   private boolean stopping = false;
   private long startingLockStamp;

   private Vertx vertx;
   private NetClient netClient;
   private NetSocket clientSocket;

   /**
    * Creates a new instance
    *
    * @param serverHost the gpsd server's hostname
    * @param serverPort the gpsd server's port
    * @param options    options for the client
    *
    * @throws NullPointerException if options is null
    */
   public GpsdClient(String serverHost, int serverPort, GpsdClientOptions options)
   {
      this.serverHost = serverHost;
      this.serverPort = serverPort;
      this.options = requireNonNull(options, "options must not be null");
   }

   /**
    * Creates a new instance
    *
    * @param serverHost the gpsd server's hostname
    * @param serverPort the gpsd server's port
    */
   public GpsdClient(String serverHost, int serverPort)
   {
      this(serverHost, serverPort, new GpsdClientOptions());
   }

   /**
    * Starts the client.
    *
    * @throws IllegalStateException if start() is called and the client is already running
    */
   public GpsdClient start()
   {
      if (isRunning()) throw new IllegalStateException("Client is already running");

      createVertx();
      createNetClient();

      connectToGpsd();

      return this;
   }

   public GpsdClient sendCommand(String rawCommand)
   {
      if (!isRunning()) throw new IllegalStateException("Client is not running");

      this.clientSocket.write(rawCommand);
      log.debug("Wrote: {}", rawCommand);

      return this;
   }

   public GpsdClient sendCommand(GpsdCommandMessage cmd)
   {
      return this.sendCommand(format(
            "?%s=%s",
            cmd.getGpsdClass(), serialize(cmd)
      ));
   }

   @SuppressWarnings("unchecked")
   public <T extends GpsdCommandMessage> GpsdClient sendCommand(T cmd, Consumer<T> consumer)
   {
      this.addHandler((Class<T>) cmd.getClass(), new Consumer<T>()
      {
         @Override
         public void accept(T t)
         {
            consumer.accept(t);
            GpsdClient.this.removeHandler(this);
         }
      });

      return this.sendCommand(cmd);
   }

   public GpsdClient watch(boolean enable, boolean dumpJson)
   {
      WatchMessage watch = new WatchMessage();
      watch.setEnable(enable);<commit_msg>Fix command result consumers sometimes executing more than once<commit_after>/*
 * Copyright 2017 <NAME> <<EMAIL>>
 *
 * 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.
 */

package com.ivkos.gpsd4j.client;

import com.ivkos.gpsd4j.messages.GpsdCommandMessage;
import com.ivkos.gpsd4j.messages.GpsdMessage;
import com.ivkos.gpsd4j.messages.WatchMessage;
import com.ivkos.gpsd4j.support.GpsdParseException;
import io.vertx.core.AsyncResult;
import io.vertx.core.Vertx;
import io.vertx.core.net.NetClient;
import io.vertx.core.net.NetClientOptions;
import io.vertx.core.net.NetSocket;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.locks.StampedLock;
import java.util.function.Consumer;

import static com.ivkos.gpsd4j.support.SerializationHelper.*;
import static java.lang.String.format;
import static java.util.Collections.*;
import static java.util.Objects.requireNonNull;

public class GpsdClient
{
   private static final Logger log = LoggerFactory.getLogger(GpsdClient.class);

   private static final int BUFFER_SIZE = 4 * 1024;

   private final Map<Class<? extends GpsdMessage>, List<Consumer<GpsdMessage>>> handlers =
         synchronizedMap(new HashMap<>());

   private final StampedLock startingLock = new StampedLock();
   private final StampedLock stoppingLock = new StampedLock();

   private final String serverHost;
   private final int serverPort;
   private final GpsdClientOptions options;

   private boolean running = false;
   private boolean stopping = false;
   private long startingLockStamp;

   private Vertx vertx;
   private NetClient netClient;
   private NetSocket clientSocket;

   /**
    * Creates a new instance
    *
    * @param serverHost the gpsd server's hostname
    * @param serverPort the gpsd server's port
    * @param options    options for the client
    *
    * @throws NullPointerException if options is null
    */
   public GpsdClient(String serverHost, int serverPort, GpsdClientOptions options)
   {
      this.serverHost = serverHost;
      this.serverPort = serverPort;
      this.options = requireNonNull(options, "options must not be null");
   }

   /**
    * Creates a new instance
    *
    * @param serverHost the gpsd server's hostname
    * @param serverPort the gpsd server's port
    */
   public GpsdClient(String serverHost, int serverPort)
   {
      this(serverHost, serverPort, new GpsdClientOptions());
   }

   /**
    * Starts the client.
    *
    * @throws IllegalStateException if start() is called and the client is already running
    */
   public GpsdClient start()
   {
      if (isRunning()) throw new IllegalStateException("Client is already running");

      createVertx();
      createNetClient();

      connectToGpsd();

      return this;
   }

   public GpsdClient sendCommand(String rawCommand)
   {
      if (!isRunning()) throw new IllegalStateException("Client is not running");

      this.clientSocket.write(rawCommand);
      log.debug("Wrote: {}", rawCommand);

      return this;
   }

   public GpsdClient sendCommand(GpsdCommandMessage cmd)
   {
      return this.sendCommand(format(
            "?%s=%s",
            cmd.getGpsdClass(), serialize(cmd)
      ));
   }

   @SuppressWarnings("unchecked")
   public <T extends GpsdCommandMessage> GpsdClient sendCommand(T cmd, Consumer<T> consumer)
   {
      this.addHandler((Class<T>) cmd.getClass(), new Consumer<T>()
      {
         // ensures this consumer doesn't get executed more than once
         private volatile boolean done = false;

         @Override
         public void accept(T t)
         {
            if (done) return;

            done = true;
            GpsdClient.this.removeHandler(this);

            consumer.accept(t);
         }
      });

      return this.sendCommand(cmd);
   }

   public GpsdClient watch(boolean enable, boolean dumpJson)
   {
      WatchMessage watch = new WatchMessage();
      watch.setEnable(enable);
{'fraction_non_alphanumeric': 0.09228339350180506, 'fraction_numerical': 0.00473826714801444, 'mean_word_length': 3.3455882352941178, 'pattern_counts': {'":': 0, '<': 27, '<?xml version=': 0, '>': 27, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '2491203', 'n_tokens_mistral': 2610, 'n_tokens_neox': 2548, 'n_words': 750}