import io.nats.client.Connection; //导入依赖的package包/类
@Test
public void testNatsConnNotClosedOnClose() throws Exception {
    try (NatsStreamingServer ignored = runServer(clusterName)) {
        // Create a NATS connection
        try (io.nats.client.Connection nc = Nats.connect()) {
            // Pass this NATS connection to NATS Streaming
            StreamingConnection sc = NatsStreaming.connect(clusterName, clientName,
                    new Options.Builder().natsConn(nc).build());
            assertNotNull(sc);
            // Now close the NATS Streaming connection
            sc.close();

            // Verify that NATS connection is not closed
            assertFalse("NATS connection should NOT have been closed in Connect",
                    nc.isClosed());
        } // nc
    } // srv
}
 

import io.nats.client.Connection; //导入依赖的package包/类
/**
 * Test method for {@link StreamingConnectionFactory#options()}.
 */
@Test
public void testOptions() throws Exception {
    StreamingConnectionFactory cf = new StreamingConnectionFactory(testClusterName, testClientName);
    cf.setAckTimeout(Duration.ofMillis(100));
    cf.setConnectTimeout(Duration.ofMillis(500));
    cf.setDiscoverPrefix("_FOO");
    cf.setMaxPubAcksInFlight(1000);
    Connection nc;

    nc = setupMockNatsConnection();

    cf.setNatsConnection(nc);
    cf.setNatsUrl("nats://foobar:1234");

    Options opts = cf.options();
    assertEquals(100, opts.getAckTimeout().toMillis());
    assertEquals(cf.getConnectTimeout(), opts.getConnectTimeout());
    assertEquals(cf.getConnectTimeout().toMillis(), opts.getConnectTimeout().toMillis());
    assertEquals(cf.getDiscoverPrefix(), opts.getDiscoverPrefix());
    assertEquals(cf.getMaxPubAcksInFlight(), opts.getMaxPubAcksInFlight());
    assertEquals(cf.getNatsUrl(), opts.getNatsUrl());
    assertEquals(cf.getNatsConnection(), opts.getNatsConn());
}
 

import io.nats.client.Connection; //导入依赖的package包/类
/**
 * A method that will publish the provided String into NATS through the defined subjects.
 * @param obj the String that will be published to NATS.
 * @throws Exception is thrown when there is no Connection nor Subject defined.
 */
@Override
protected void publishToNats(byte[] payload) throws Exception {
	resetClosingTimeout();
	
	final Message natsMessage = new Message();

	natsMessage.setData(payload, 0, payload.length);

	final Connection localConnection = getConnection();
	for (String subject : getDefinedSubjects()) {
		natsMessage.setSubject(subject);
		localConnection.publish(natsMessage);

		logger.trace("Send '{}' from Spark to NATS ({})", payload, subject);
	}
}
 

import io.nats.client.Connection; //导入依赖的package包/类
public void run() throws Exception {
    try (Connection nc = Nats.connect(url)) {
        Message msg = nc.request(subject, payload.getBytes(), 100, TimeUnit.MILLISECONDS);
        System.out.printf("Published [%s] : '%s'\n", subject, payload);
        if (msg == null) {
            Exception err = nc.getLastException();
            if (err != null) {
                System.err.println("Error in request");
                throw err;
            }
            throw new TimeoutException("Request timed out");
        }
        System.out.printf("Received [%s] : '%s'\n", msg.getSubject(),
                msg.getData() == null ? "" : new String(msg.getData()));
    }
}
 

import io.nats.client.Connection; //导入依赖的package包/类
@Test
public void testMaxPubAcksInFlight() throws Exception {
    try (NatsStreamingServer srv = runServer(clusterName)) {
        try (Connection nc = Nats.connect()) {
            Options opts = new Options.Builder()
                    .maxPubAcksInFlight(1)
                    .pubAckWait(Duration.ofSeconds(1))
                    .natsConn(nc)
                    .build();

            StreamingConnection sc = NatsStreaming.connect(clusterName, clientName, opts);
            // Don't defer the close of connection since the server is stopped,
            // the close would delay the test.

            // Cause the ACK to not come by shutdown the server now
            srv.shutdown();

            byte[] msg = "hello".getBytes();

            // Send more than one message, if MaxPubAcksInflight() works, one
            // of the publish call should block for up to PubAckWait.
            Instant start = Instant.now();
            for (int i = 0; i < 2; i++) {
                sc.publish("foo", msg, null);
            }
            Instant end = Instant.now();
            // So if the loop ended before the PubAckWait timeout, then it's a failure.
            if (Duration.between(start, end).compareTo(Duration.ofSeconds(1)) < 0) {
                fail("Should have blocked after 1 message sent");
            }
        }
    }
}
 

import io.nats.client.Connection; //导入依赖的package包/类
@Test
public void testNatsConnectionName() throws Exception {
    try (NatsStreamingServer ignored = runServer(clusterName)) {
        Options opts = new Options.Builder().build();
        try (StreamingConnection sc = NatsStreaming.connect(clusterName, clientName, opts)) {
            Connection nc = sc.getNatsConnection();
            assertEquals(clientName, nc.getName());
        }
    }
}
 

import io.nats.client.Connection; //导入依赖的package包/类
/**
 * Test method for {@link StreamingConnectionFactory#createConnection()}.
 */
@Test
public void testCreateConnection() throws Exception {
    try (io.nats.client.Connection nc = setupMockNatsConnection()) {
        StreamingConnectionFactory cf = new StreamingConnectionFactory(testClusterName, testClientName);
        cf.setNatsConnection(nc);
        try (StreamingConnection sc = cf.createConnection()) {
            assertTrue(sc instanceof StreamingConnectionImpl);
        } catch (IOException | TimeoutException e) {
            e.printStackTrace();
            fail(e.getMessage());
        }
    }
}
 

import io.nats.client.Connection; //导入依赖的package包/类
/**
 * Test method for {@link java.io.Serializable}.
 * @throws IOException 
 * @throws ClassNotFoundException 
 * @throws InterruptedException 
 */
@Test
public void testSerializable() throws ClassNotFoundException, IOException, InterruptedException {
	Connection nc = setupMockNatsConnection();
	Options.Builder testOptsBuilder = new Options.Builder()
			.pubAckWait(Duration.ofMillis(500))
			.connectWait(Duration.ofMillis(1500))
			.discoverPrefix("PrEfiX")
			.maxPubAcksInFlight(10000)
			.natsConn(nc)
			.natsUrl("nats://nats");
	final Options.Builder serializedTestOpts = (Options.Builder) UnitTestUtilities.serializeDeserialize(testOptsBuilder);
	assertTrue(equals(testOptsBuilder, serializedTestOpts));
}
 

import io.nats.client.Connection; //导入依赖的package包/类
/** Create a socket connection and receive data until receiver is stopped 
 * @throws IncompleteException 
 * @throws TimeoutException 
 * @throws IOException 
 * @throws Exception **/
protected void receive() throws IncompleteException, IOException, TimeoutException {

	// Make connection and initialize streams			  
	final ConnectionFactory connectionFactory = new ConnectionFactory(getEnrichedProperties());
	final Connection connection = connectionFactory.createConnection();
	logger.info("A NATS from '{}' to Spark Connection has been created for '{}', sharing Queue '{}'.", connection.getConnectedUrl(), this, queue);
	
	for (String subject: getSubjects()) {
		final Subscription sub = connection.subscribe(subject, queue, getMessageHandler());
		logger.info("Listening on {}.", subject);
		
		Runtime.getRuntime().addShutdownHook(new Thread(new Runnable(){
			@Override
			public void run() {
				logger.debug("Caught CTRL-C, shutting down gracefully..." + this);
				try {
					sub.unsubscribe();
				} catch (IOException e) {
					if (logger.isDebugEnabled()) {
						logger.error("Exception while unsubscribing " + e.toString());
					}
				}
				connection.close();
			}
		}));
	}
}
 

import io.nats.client.Connection; //导入依赖的package包/类
protected Connection createConnection() throws IOException, TimeoutException, Exception {
	final Connection newConnection = getConnectionFactory().createConnection();
	logger.debug("A NATS Connection {} has been created for {}", newConnection, this);
	
	Runtime.getRuntime().addShutdownHook(new Thread(new Runnable(){
		@Override
		public void run() {
			logger.debug("Caught CTRL-C, shutting down gracefully... " + this);
			newConnection.close();
		}
	}));
	return newConnection;
}
 

import io.nats.client.Connection; //导入依赖的package包/类
@Test
public void testLoadConsumer() throws InterruptedException, IOException, TimeoutException {
    mockResultEndpoint.setExpectedMessageCount(10000);
    ConnectionFactory cf = new ConnectionFactory("nats://localhost:4222");
    Connection connection = cf.createConnection();

    for (int i = 0; i < 10000; i++) {
        connection.publish("test", ("test" + i).getBytes());
    }

    mockResultEndpoint.assertIsSatisfied();
}
 

import io.nats.client.Connection; //导入依赖的package包/类
/**
 * Constructs a new benchmark sample.
 *
 * @param jobCount the number of messages that were sent or received
 * @param msgSize  the size in bytes of each messages
 * @param start    the start time in nanoseconds
 * @param end      the end time in nanoseconds
 * @param nc       the NATs connection
 */
public Sample(int jobCount, int msgSize, long start, long end, Connection nc) {
    this.jobMsgCnt = jobCount;
    this.start = start;
    this.end = end;
    this.msgBytes = (long) msgSize * jobCount;
    Statistics stats = nc.getStats();
    this.msgCnt = stats.getOutMsgs() + stats.getInMsgs();
    this.ioBytes = stats.getOutBytes() + stats.getInBytes();
}
 

import io.nats.client.Connection; //导入依赖的package包/类
/**
 * Returns a million message sample.
 *
 * @return a Sample for one million messages
 */
private Sample millionMessagesSecondSample(int seconds) {
    int messages = MILLION * seconds;
    long start = baseTime;
    long end = start + TimeUnit.SECONDS.toNanos(seconds);

    final Connection nc = mock(Connection.class);
    when(nc.getStats()).thenReturn(new Statistics());
    Sample stat = new Sample(messages, MSG_SIZE, start, end, nc);
    stat.msgCnt = (long) messages;
    stat.msgBytes = (long) messages * MSG_SIZE;
    stat.ioBytes = stat.msgBytes;
    return stat;
}
 

import io.nats.client.Connection; //导入依赖的package包/类
static synchronized Connection newDefaultConnection() throws IOException, TimeoutException {
    return new ConnectionFactory().createConnection();
}
 

import io.nats.client.Connection; //导入依赖的package包/类
protected synchronized Connection getConnection() throws Exception {
	if (connection == null) {
		connection = createConnection();
	}
	return connection;
}
 

import io.nats.client.Connection; //导入依赖的package包/类
private Connection getConnection() throws TimeoutException, IOException {
    Properties prop = getEndpoint().getNatsConfiguration().createProperties();
    ConnectionFactory factory = new ConnectionFactory(prop);
    connection = factory.createConnection();
    return connection;
}
 

import io.nats.client.Connection; //导入依赖的package包/类
private Connection getConnection() throws IOException, InterruptedException, TimeoutException {
    Properties prop = getEndpoint().getNatsConfiguration().createProperties();
    ConnectionFactory factory = new ConnectionFactory(prop);
    connection = factory.createConnection();
    return connection;
}
 

import io.nats.client.Connection; //导入依赖的package包/类
NatsConsumingTask(Connection connection, NatsConfiguration configuration) {
    this.connection = connection;
    this.configuration = configuration;
}
 

import io.nats.client.Connection; //导入依赖的package包/类
public void run() throws IOException {
    try (Connection nc = Nats.connect(url)) {
        nc.publish(subject, payload.getBytes());
        System.out.printf("Published [%s] : '%s'\n", subject, payload);
    }
}
 

import io.nats.client.Connection; //导入依赖的package包/类
/**
 * Returns the NATS StreamingConnection, if set.
 *
 * @return the NATS StreamingConnection
 */
public io.nats.client.Connection getNatsConnection() {
    return this.natsConn;
}