def go(loop):
    nc = NATS()

    try:
        yield from nc.connect(io_loop=loop)
    except:
        pass

    @asyncio.coroutine
    def message_handler(msg):
        print("[Received on '{}']: {}".format(msg.subject, msg.data.decode()))

    try:
        # Interested in receiving 2 messages from the 'discover' subject.
        sid = yield from nc.subscribe("discover", "", message_handler)
        yield from nc.auto_unsubscribe(sid, 2)

        yield from nc.publish("discover", b'hello')
        yield from nc.publish("discover", b'world')

        # Following 2 messages won't be received.
        yield from nc.publish("discover", b'again')
        yield from nc.publish("discover", b'!!!!!')
    except ErrConnectionClosed:
        print("Connection closed prematurely")

    @asyncio.coroutine
    def request_handler(msg):
        print("[Request on '{} {}']: {}".format(msg.subject, msg.reply, msg.data.decode()))
        yield from nc.publish(msg.reply, b'OK')

    if nc.is_connected:
        
        # Subscription using a 'workers' queue so that only a single subscriber
        # gets a request at a time.
        yield from nc.subscribe("help", "workers", cb=request_handler)

        try:
            # Make a request expecting a single response within 500 ms,
            # otherwise raising a timeout error.
            msg = yield from nc.timed_request("help", b'help please', 0.500)
            print("[Response]: {}".format(msg.data))

            # Make a roundtrip to the server to ensure messages
            # that sent messages have been processed already.
            yield from nc.flush(0.500)
        except ErrTimeout:
            print("[Error] Timeout!")

        # Wait a bit for message to be dispatched...
        yield from asyncio.sleep(1, loop=loop)

        # Detach from the server.
        yield from nc.close()

    if nc.last_error is not None:
        print("Last Error: {}".format(nc.last_error))

    if nc.is_closed:
        print("Disconnected.")

def main(loop):
    parser = argparse.ArgumentParser()
    parser.add_argument('-n', '--count', default=DEFAULT_NUM_MSGS, type=int)
    parser.add_argument('-s', '--size', default=DEFAULT_MSG_SIZE, type=int)
    parser.add_argument('-S', '--subject', default='test')
    parser.add_argument('-b', '--batch', default=DEFAULT_BATCH_SIZE, type=int)
    parser.add_argument('--servers', default=[], action='append')
    args = parser.parse_args()

    data = []
    for i in range(0, args.size):
        s = "%01x" % randint(0, 15)
        data.append(s.encode())
    payload = b''.join(data)

    servers = args.servers
    if len(args.servers) < 1:
        servers = ["nats://127.0.0.1:4222"]
    opts = { "servers": servers, "io_loop": loop }

    # Make sure we're connected to a server first..
    nc = NATS()
    try:
        yield from nc.connect(**opts)
    except Exception as e:
        sys.stderr.write("ERROR: {0}".format(e))
        show_usage_and_die()

    # Start the benchmark
    start = time.time()
    to_send = args.count

    print("Sending {0} messages of size {1} bytes on [{2}]".format(
        args.count, args.size, args.subject))
    while to_send > 0:
        for i in range(0, args.batch):
            to_send -= 1
            yield from nc.publish(args.subject, payload)
            if (to_send % HASH_MODULO) == 0:
                sys.stdout.write("#")
                sys.stdout.flush()
            if to_send == 0:
                break

        # Minimal pause in between batches sent to server
        yield from asyncio.sleep(0.00001, loop=loop)

    # Additional roundtrip with server to try to ensure everything has been sent already.
    try:
        yield from nc.flush(DEFAULT_FLUSH_TIMEOUT)
    except ErrTimeout:
        print("Server flush timeout after {0}".format(DEFAULT_FLUSH_TIMEOUT))

    elapsed = time.time() - start
    mbytes = "%.1f" % (((args.size * args.count)/elapsed) / (1024*1024))
    print("\nTest completed : {0} msgs/sec ({1}) MB/sec".format(
        args.count/elapsed,
        mbytes))
    yield from nc.close()

  def test_invalid_subscription_type(self):
    nc = NATS()

    with self.assertRaises(NatsError):
      yield from nc.subscribe("hello", cb=None, future=None)

    with self.assertRaises(NatsError):
      yield from nc.subscribe_async("hello", cb=None)

 def test_flush(self):
   nc = NATS()
   yield from nc.connect(io_loop=self.loop)
   for i in range(0, 10):
     yield from nc.publish("flush.%d" % i, b'AA')
     yield from nc.flush()
   self.assertEqual(10, nc.stats['out_msgs'])
   self.assertEqual(20, nc.stats['out_bytes'])
   yield from nc.close()

 def test_default_connect(self):
   nc = NATS()
   yield from nc.connect(io_loop=self.loop)
   self.assertIn('auth_required', nc._server_info)
   self.assertIn('max_payload', nc._server_info)
   self.assertEqual(nc._server_info['max_payload'], nc._max_payload)
   self.assertTrue(nc.is_connected)
   yield from nc.close()
   self.assertTrue(nc.is_closed)
   self.assertFalse(nc.is_connected)

 def test_pending_data_size_tracking(self):
   nc = NATS()
   yield from nc.connect(io_loop=self.loop)
   largest_pending_data_size = 0
   for i in range(0,100):
     yield from nc.publish("example", b'A' * 100000)
     if nc.pending_data_size > 0:
       largest_pending_data_size = nc.pending_data_size
   self.assertTrue(largest_pending_data_size > 0)
   yield from nc.close()

    def test_pending_data_size_flush_on_close(self):
        nc = NATS()

        disconnected_count = 0
        reconnected_count = 0
        closed_count = 0
        err_count = 0

        @asyncio.coroutine
        def disconnected_cb():
            nonlocal disconnected_count
            disconnected_count += 1

        @asyncio.coroutine
        def reconnected_cb():
            nonlocal reconnected_count
            reconnected_count += 1

        @asyncio.coroutine
        def closed_cb():
            nonlocal closed_count
            closed_count += 1

        options = {
            'dont_randomize': True,
            'io_loop': self.loop,
            'disconnected_cb': disconnected_cb,
            'closed_cb': closed_cb,
            'reconnected_cb': reconnected_cb,
            'reconnect_time_wait': 0.01
        }
        yield from nc.connect(**options)

        total_received = 0
        future = asyncio.Future(loop=self.loop)

        @asyncio.coroutine
        def receiver_cb(msg):
            nonlocal total_received
            total_received += 1
            if total_received == 200:
                future.set_result(True)

        # Extra connection which should be receiving all the messages
        nc2 = NATS()
        yield from nc2.connect(**options)
        yield from nc2.subscribe("example.*", cb=receiver_cb)
        yield from nc2.flush()

        for i in range(0, 200):
            yield from nc.publish("example.{}".format(i), b'A' * 20)

        # All pending messages should have been emitted to the server
        # by the first connection at this point.
        yield from nc.close()

        # Wait for the server to flush all the messages back to the receiving client
        yield from asyncio.wait_for(future, 1, loop=self.loop)
        yield from nc2.close()
        self.assertEqual(total_received, 200)

def run(loop):
    parser = argparse.ArgumentParser()

    # e.g. nats-sub hello -s nats://127.0.0.1:4222
    parser.add_argument('subject', default='hello', nargs='?')
    parser.add_argument('-s', '--servers', default=[], action='append')
    parser.add_argument('-q', '--queue', default="")
    args = parser.parse_args()

    nc = NATS()

    @asyncio.coroutine
    def closed_cb():
        print("Connection to NATS is closed.")
        yield from asyncio.sleep(0.1, loop=loop)
        loop.stop()

    @asyncio.coroutine
    def reconnected_cb():
        print("Connected to NATS at {}...".format(nc.connected_url.netloc))

    @asyncio.coroutine
    def subscribe_handler(msg):
        subject = msg.subject
        reply = msg.reply
        data = msg.data.decode()
        print("Received a message on '{subject} {reply}': {data}".format(
          subject=subject, reply=reply, data=data))

    options = {
        "io_loop": loop,
        "closed_cb": closed_cb,
        "reconnected_cb": reconnected_cb
    }

    try:
        if len(args.servers) > 0:
            options['servers'] = args.servers

        yield from nc.connect(**options)
    except Exception as e:
        print(e)
        show_usage_and_die()

    print("Connected to NATS at {}...".format(nc.connected_url.netloc))
    def signal_handler():
        if nc.is_closed:
            return
        print("Disconnecting...")
        loop.create_task(nc.close())

    for sig in ('SIGINT', 'SIGTERM'):
        loop.add_signal_handler(getattr(signal, sig), signal_handler)

    yield from nc.subscribe(args.subject, args.queue, subscribe_handler)

def main(loop):
    parser = argparse.ArgumentParser()
    parser.add_argument('-n', '--count', default=DEFAULT_NUM_MSGS, type=int)
    parser.add_argument('-S', '--subject', default='test')
    parser.add_argument('--servers', default=[], action='append')
    args = parser.parse_args()

    servers = args.servers
    if len(args.servers) < 1:
        servers = ["nats://127.0.0.1:4222"]
    opts = { "servers": servers, "io_loop": loop, "allow_reconnect": False }

    # Make sure we're connected to a server first...
    nc = NATS()
    try:
        yield from nc.connect(**opts)
    except Exception as e:
        sys.stderr.write("ERROR: {0}".format(e))
        show_usage_and_die()

    received = 0
    start = None

    @asyncio.coroutine
    def handler(msg):
        nonlocal received
        nonlocal start
        received += 1

        # Measure time from when we get the first message.
        if received == 1:
            start = time.monotonic()
        if (received % HASH_MODULO) == 0:
            sys.stdout.write("*")
            sys.stdout.flush()

    yield from nc.subscribe(args.subject, cb=handler)

    # Additional roundtrip with server to ensure everything has been
    # processed by the server already.
    yield from nc.flush()

    print("Waiting for {} messages on [{}]...".format(args.count, args.subject))
    try:
        while received < args.count:
            yield from asyncio.sleep(0.1, loop=loop)
    except ErrTimeout:
        print("Server flush timeout after {0}".format(DEFAULT_FLUSH_TIMEOUT))

    elapsed = time.monotonic() - start
    print("\nTest completed : {0} msgs/sec sent".format(args.count/elapsed))

    print("Received {0} messages ({1} msgs/sec)".format(received, received/elapsed))
    yield from nc.close()

 def test_connect(self):
     nc = NATS()
     yield from nc.connect(io_loop=self.loop, servers=['nats://localhost:4224'],
                           tls=self.ssl_ctx)
     self.assertEqual(nc._server_info['max_payload'], nc.max_payload)
     self.assertTrue(nc._server_info['tls_required'])
     self.assertTrue(nc._server_info['tls_verify'])
     self.assertTrue(nc.max_payload > 0)
     self.assertTrue(nc.is_connected)
     yield from nc.close()
     self.assertTrue(nc.is_closed)
     self.assertFalse(nc.is_connected)

  def test_timed_request(self):
    nc = NATS()
    msgs = []
    counter = 0

    @asyncio.coroutine
    def worker_handler(msg):
      nonlocal counter
      counter += 1
      msgs.append(msg)
      yield from nc.publish(msg.reply, 'Reply:{}'.format(counter).encode())

    @asyncio.coroutine
    def slow_worker_handler(msg):
      yield from asyncio.sleep(0.5, loop=self.loop)
      yield from nc.publish(msg.reply, b'timeout by now...')

    yield from nc.connect(io_loop=self.loop)
    yield from nc.subscribe("help", cb=worker_handler)
    yield from nc.subscribe("slow.help", cb=slow_worker_handler)

    response = yield from nc.timed_request("help", b'please', timeout=1)
    self.assertEqual(b'Reply:1', response.data)
    response = yield from nc.timed_request("help", b'please', timeout=1)
    self.assertEqual(b'Reply:2', response.data)

    with self.assertRaises(ErrTimeout):
      yield from nc.timed_request("slow.help", b'please', timeout=0.1)
    yield from asyncio.sleep(1, loop=self.loop)
    yield from nc.close()

    def test_connect_with_auth_token(self):
        nc = NATS()

        options = {
            'servers': [
                "nats://[email protected]:4223",
            ],
            'io_loop': self.loop
        }
        yield from nc.connect(**options)
        self.assertIn('auth_required', nc._server_info)
        self.assertTrue(nc.is_connected)
        yield from nc.close()
        self.assertTrue(nc.is_closed)
        self.assertFalse(nc.is_connected)

  def test_subscribe_async_without_coroutine_unsupported(self):
    nc = NATS()
    msgs = []

    def subscription_handler(msg):
      if msg.subject == "tests.1":
        time.sleep(0.5)
      if msg.subject == "tests.3":
        time.sleep(0.2)
      msgs.append(msg)

    yield from nc.connect(io_loop=self.loop)

    with self.assertRaises(NatsError):
      sid = yield from nc.subscribe_async("tests.>", cb=subscription_handler)
    yield from nc.close()

def run(loop):
    parser = argparse.ArgumentParser()

    # e.g. nats-pub hello -d "world" -s nats://127.0.0.1:4222 -s nats://127.0.0.1:4223
    parser.add_argument('subject', default='hello', nargs='?')
    parser.add_argument('-d', '--data', default="hello world")
    parser.add_argument('-s', '--servers', default=[], action='append')
    args = parser.parse_args()

    nc = NATS()

    @asyncio.coroutine
    def closed_cb():
        print("Connection to NATS is closed.")
        loop.stop()

    @asyncio.coroutine
    def reconnected_cb():
        print("Connected to NATS at {}...".format(nc.connected_url.netloc))

    @asyncio.coroutine
    def subscribe_handler(msg):
        subject = msg.subject
        reply = msg.reply
        data = msg.data.decode()
        print("Received a message on '{subject} {reply}': {data}".format(
          subject=subject, reply=reply, data=data))

    options = {
        "io_loop": loop,
        "closed_cb": closed_cb,
        "reconnected_cb": reconnected_cb
    }

    try:
        if len(args.servers) > 0:
            options['servers'] = args.servers

        yield from nc.connect(**options)
    except Exception as e:
        print(e)
        show_usage_and_die()

    print("Connected to NATS at {}...".format(nc.connected_url.netloc))
    yield from nc.publish(args.subject, args.data.encode())
    yield from nc.flush()
    yield from nc.close()

def run(loop):
    nc = NATS()

    ssl_ctx = ssl.create_default_context(purpose=ssl.Purpose.SERVER_AUTH)
    ssl_ctx.protocol = ssl.PROTOCOL_TLSv1_2
    ssl_ctx.load_verify_locations('../tests/certs/ca.pem')
    ssl_ctx.load_cert_chain(certfile='../tests/certs/client-cert.pem',
                            keyfile='../tests/certs/client-key.pem')
    yield from nc.connect(io_loop=loop, tls=ssl_ctx)

    @asyncio.coroutine
    def message_handler(msg):
        subject = msg.subject
        reply = msg.reply
        data = msg.data.decode()
        print("Received a message on '{subject} {reply}': {data}".format(
            subject=subject, reply=reply, data=data))

    # Simple publisher and async subscriber via coroutine.
    sid = yield from nc.subscribe("foo", cb=message_handler)

    # Stop receiving after 2 messages.
    yield from nc.auto_unsubscribe(sid, 2)
    yield from nc.publish("foo", b'Hello')
    yield from nc.publish("foo", b'World')
    yield from nc.publish("foo", b'!!!!!')

    @asyncio.coroutine
    def help_request(msg):
        subject = msg.subject
        reply = msg.reply
        data = msg.data.decode()
        print("Received a message on '{subject} {reply}': {data}".format(
            subject=subject, reply=reply, data=data))
        yield from nc.publish(reply, b'I can help')

    # Use queue named 'workers' for distributing requests
    # among subscribers.
    yield from nc.subscribe("help", "workers", help_request)

    # Send a request and expect a single response
    # and trigger timeout if not faster than 50 ms.
    try:
        response = yield from nc.timed_request("help", b'help me', 0.050)
        print("Received response: {message}".format(message=response.data.decode()))
    except ErrTimeout:
        print("Request timed out")

    yield from asyncio.sleep(1, loop=loop)
    yield from nc.close()

def main(loop):
  parser = argparse.ArgumentParser()
  parser.add_argument('-n', '--iterations', default=DEFAULT_ITERATIONS, type=int)
  parser.add_argument('-S', '--subject', default='test')
  parser.add_argument('--servers', default=[], action='append')
  args = parser.parse_args()

  servers = args.servers
  if len(args.servers) < 1:
    servers = ["nats://127.0.0.1:4222"]
  opts = { "servers": servers }

  # Make sure we're connected to a server first...
  nc = NATS()
  try:
    yield from nc.connect(**opts)
  except Exception as e:
    sys.stderr.write("ERROR: {0}".format(e))
    show_usage_and_die()

  @asyncio.coroutine
  def handler(msg):
    yield from nc.publish(msg.reply, b'')
  yield from nc.subscribe(args.subject, cb=handler)

  # Start the benchmark
  start = time.monotonic()
  to_send = args.iterations

  print("Sending {0} request/responses on [{1}]".format(
      args.iterations, args.subject))
  while to_send > 0:
    to_send -= 1
    if to_send == 0:
      break

    yield from nc.timed_request(args.subject, b'')
    if (to_send % HASH_MODULO) == 0:
      sys.stdout.write("#")
      sys.stdout.flush()

  duration = time.monotonic() - start
  ms = "%.3f" % ((duration/args.iterations) * 1000)
  print("\nTest completed : {0} ms avg request/response latency".format(ms))
  yield from nc.close()

def run(loop):
    nc = Nats()
    yield from nc.connect(io_loop=loop)

    # Send a request and expect a single response and trigger timeout if not
    # faster than 50 ms.
    try:
        response = yield from nc.timed_request("conf.host", b'host', 0.050)
        print("Received response: {message}".format(message=response.data.decode()))
    except ErrTimeout:
        print("Request timed out")

    yield from nc.publish("log.info", b'initializing')
    yield from nc.publish("log.info", b'scraping item 1')

    @asyncio.coroutine
    def help_request(msg):
        subject = msg.subject
        reply = msg.reply
        data = msg.data.decode()
        print("Received a message on '{subject} {reply}': {data}".format(
            subject=subject, reply=reply, data=data
)       )
        yield from nc.publish(reply, b'I can help')

    # Use queue named 'workers' for distributing requests among subscribers.
    yield from nc.subscribe("cmd.help", "workers", help_request)

    yield from asyncio.sleep(20, loop=loop)
    yield from nc.close()

    def test_connect_with_bad_auth_token(self):
        nc = NATS()

        options = {
            'servers': [
                "nats://[email protected]:4225",
            ],
            'allow_reconnect': False,
            'reconnect_time_wait': 0.1,
            'max_reconnect_attempts': 1,
            'io_loop': self.loop
        }
        # Authorization Violation
        with self.assertRaises(NatsError):
            yield from nc.connect(**options)

        self.assertIn('auth_required', nc._server_info)
        self.assertFalse(nc.is_connected)

  def test_connect_with_auth(self):
    nc = NATS()

    options = {
      'servers': [
        "nats://foo:[email protected]:4223",
        "nats://hoge:[email protected]:4224"
        ],
      'io_loop': self.loop
      }
    yield from nc.connect(**options)
    self.assertIn('auth_required', nc._server_info)
    self.assertIn('max_payload', nc._server_info)
    self.assertEqual(nc._server_info['max_payload'], nc._max_payload)
    self.assertTrue(nc.is_connected)
    yield from nc.close()
    self.assertTrue(nc.is_closed)
    self.assertFalse(nc.is_connected)

  def test_connect_with_failed_auth(self):
    nc = NATS()

    options = {
      'servers': [
        "nats://hello:[email protected]:4223",
        ],
      'io_loop': self.loop
      }
    with self.assertRaises(ErrNoServers):
      yield from nc.connect(**options)

    self.assertIn('auth_required', nc._server_info)
    self.assertTrue(nc._server_info['auth_required'])
    self.assertFalse(nc.is_connected)
    yield from nc.close()
    self.assertTrue(nc.is_closed)
    self.assertEqual(ErrNoServers, type(nc.last_error))
    self.assertEqual(0, nc.stats['reconnects'])