<commit_before>/*******************************************************************************
 * Copyright (c) 2015-2016 Apcera Inc.
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the MIT License (MIT)
 * which accompanies this distribution, and is available at
 * http://opensource.org/licenses/MIT
 *******************************************************************************/
package io.nats.client.examples;
import java.io.IOException;
import java.nio.charset.Charset;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import io.nats.client.*;

public class Replier {
	Map<String, String> parsedArgs = new HashMap<String, String>();

	int count = 20000;
	String url = ConnectionFactory.DEFAULT_URL;
	String subject = "foo";
	boolean sync = false;
	int received = 0;
	boolean verbose = false;

	long start = 0L;
	long end = 0L;

	Lock testLock = new ReentrantLock();
	Condition allDone = testLock.newCondition();
	boolean done = false;

	byte[] replyBytes = "reply".getBytes(Charset.forName("UTF-8"));

	public void run(String[] args)
	{
		parseArgs(args);
		banner();

		try (Connection c = new ConnectionFactory(url).createConnection()) {
			long elapsed, seconds;

			if (sync)
			{
				elapsed = receiveSyncSubscriber(c);
			}
			else
			{
				elapsed = receiveAsyncSubscriber(c);
			}
			seconds = TimeUnit.NANOSECONDS.toSeconds(elapsed);
			System.out.printf("Replied to %d msgs in %d seconds ", received, seconds);

			if (seconds > 0) {
				System.out.printf("(%d msgs/second).\n",
						(received / seconds));
			} else {
				System.out.println();
				System.out.println("Test not long enough to produce meaningful stats. "
						+ "Please increase the message count (-count n)");
			}
			printStats(c);
		} catch (IOException | TimeoutException e) {
			System.err.println("Couldn't connect: " + e.getMessage());
			System.exit(-1);
		}
	}

	private void printStats(Connection c)
	{
		Statistics s = c.getStats();
		System.out.printf("Statistics:  \n");
		System.out.printf("   Incoming Payload Bytes: %d\n", s.getInBytes());
		System.out.printf("   Incoming Messages: %d\n", s.getInMsgs());
		System.out.printf("   Outgoing Payload Bytes: %d\n", s.getOutBytes());
		System.out.printf("   Outgoing Messages: %d\n", s.getOutMsgs());
	}


	private long receiveAsyncSubscriber(Connection conn)
	{
		final Connection c = conn;
		MessageHandler mh = new MessageHandler() {
			public void onMessage(Message msg) {
				if (received++ == 0)
					start = System.nanoTime();

				if (verbose)
					System.out.println("Received: " + msg);

				try {
					c.publish(msg.getReplyTo(), replyBytes);
				} catch (IllegalStateException e) {
					e.printStackTrace();
				}

				if (received >= count)
				{
					end = System.nanoTime();
					testLock.lock();
					try
					{
						done=true;
						allDone.signal();
					}
					finally {
						testLock.unlock();
					}
				}		
			}
		};
		
		AsyncSubscription s = c.subscribeAsync(subject, mh);
		// just wait to complete
		testLock.lock();
		try
		{
			s.start();
			while (!done)<commit_msg>Updated to reflect possible IOException<commit_after>/*******************************************************************************
 * Copyright (c) 2015-2016 Apcera Inc.
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the MIT License (MIT)
 * which accompanies this distribution, and is available at
 * http://opensource.org/licenses/MIT
 *******************************************************************************/
package io.nats.client.examples;
import java.io.IOException;
import java.nio.charset.Charset;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import io.nats.client.*;

public class Replier {
	Map<String, String> parsedArgs = new HashMap<String, String>();

	int count = 20000;
	String url = ConnectionFactory.DEFAULT_URL;
	String subject = "foo";
	boolean sync = false;
	int received = 0;
	boolean verbose = false;

	long start = 0L;
	long end = 0L;

	Lock testLock = new ReentrantLock();
	Condition allDone = testLock.newCondition();
	boolean done = false;

	byte[] replyBytes = "reply".getBytes(Charset.forName("UTF-8"));

	public void run(String[] args)
	{
		parseArgs(args);
		banner();

		try (Connection c = new ConnectionFactory(url).createConnection()) {
			long elapsed, seconds;

			if (sync)
			{
				elapsed = receiveSyncSubscriber(c);
			}
			else
			{
				elapsed = receiveAsyncSubscriber(c);
			}
			seconds = TimeUnit.NANOSECONDS.toSeconds(elapsed);
			System.out.printf("Replied to %d msgs in %d seconds ", received, seconds);

			if (seconds > 0) {
				System.out.printf("(%d msgs/second).\n",
						(received / seconds));
			} else {
				System.out.println();
				System.out.println("Test not long enough to produce meaningful stats. "
						+ "Please increase the message count (-count n)");
			}
			printStats(c);
		} catch (IOException | TimeoutException e) {
			System.err.println("Couldn't connect: " + e.getMessage());
			System.exit(-1);
		}
	}

	private void printStats(Connection c)
	{
		Statistics s = c.getStats();
		System.out.printf("Statistics:  \n");
		System.out.printf("   Incoming Payload Bytes: %d\n", s.getInBytes());
		System.out.printf("   Incoming Messages: %d\n", s.getInMsgs());
		System.out.printf("   Outgoing Payload Bytes: %d\n", s.getOutBytes());
		System.out.printf("   Outgoing Messages: %d\n", s.getOutMsgs());
	}


	private long receiveAsyncSubscriber(Connection conn)
	{
		final Connection c = conn;
		MessageHandler mh = new MessageHandler() {
			public void onMessage(Message msg) {
				if (received++ == 0)
					start = System.nanoTime();

				if (verbose)
					System.out.println("Received: " + msg);

				try {
					c.publish(msg.getReplyTo(), replyBytes);
				} catch (IllegalStateException | IOException e) {
					e.printStackTrace();
				}

				if (received >= count)
				{
					end = System.nanoTime();
					testLock.lock();
					try
					{
						done=true;
						allDone.signal();
					}
					finally {
						testLock.unlock();
					}
				}		
			}
		};
		
		AsyncSubscription s = c.subscribeAsync(subject, mh);
		// just wait to complete
		testLock.lock();
		try
		{
			s.start();
			while (!done)
{'fraction_non_alphanumeric': 0.1750897129186603, 'fraction_numerical': 0.005980861244019139, 'mean_word_length': 3.6035788024776325, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 11, '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': '1450878', 'n_tokens_mistral': 2455, 'n_tokens_neox': 2025, 'n_words': 514}