@Test
public void testUrlEncodingAndDecoding()
    throws TestFailure, ExecutionException, TimeoutException, InterruptedException {

  DatabaseReference ref = FirebaseDatabase.getInstance(masterApp)
      .getReference("/a%b&[email protected]/space: /non-ascii:ø");
  String result = ref.toString();
  String expected = IntegrationTestUtils.getDatabaseUrl()
      + "/a%25b%26c%40d/space%3A%20/non-ascii%3A%C3%B8";
  assertEquals(expected, result);

  String child = "" + new Random().nextInt(100000000);
  new WriteFuture(ref.child(child), "testdata").timedGet();
  DataSnapshot snap = TestHelpers.getSnap(ref.child(child));
  assertEquals("testdata", snap.getValue());
}
 

@SuppressWarnings("unchecked")
@Test
public void shouldGetItem() throws InterruptedException, TimeoutException, ExecutionException {
    // Given
    final String key = Randoms.randomString();
    final String item = Randoms.randomString();
    final int timeout = Randoms.randomInt(5) + 1;
    final GetFuture<Object> f = mock(GetFuture.class);
    when(memcachedClient.asyncGet(key)).thenReturn(f);
    when(f.get(timeout, TimeUnit.SECONDS)).thenReturn(item);

    // When
    final Object obj = memcachedItemCache.getItem(key, timeout);

    // Then
    assertEquals(item, obj);
}
 

@Test
public void testGetPriority()
    throws TimeoutException, InterruptedException, TestFailure {
  DatabaseReference ref = IntegrationTestUtils.getRandomNode(masterApp);

  ReadFuture readFuture = ReadFuture.untilCountAfterNull(ref, 7);

  ref.setValueAsync("a");
  ref.setValueAsync("b", 5);
  ref.setValueAsync("c", "6");
  ref.setValueAsync("d", 7);
  ref.setValueAsync(new MapBuilder().put(".value", "e").put(".priority", 8).build());
  ref.setValueAsync(new MapBuilder().put(".value", "f").put(".priority", "8").build());
  ref.setValueAsync(new MapBuilder().put(".value", "g").put(".priority", null).build());

  List<EventRecord> events = readFuture.timedGet();
  assertNull(events.get(0).getSnapshot().getPriority());
  assertEquals(5.0, events.get(1).getSnapshot().getPriority());
  assertEquals("6", events.get(2).getSnapshot().getPriority());
  assertEquals(7.0, events.get(3).getSnapshot().getPriority());
  assertEquals(8.0, events.get(4).getSnapshot().getPriority());
  assertEquals("8", events.get(5).getSnapshot().getPriority());
  assertNull(events.get(6).getSnapshot().getPriority());
}
 

@Test
public void nextWithTimeout() throws Exception {
    BlockingIterator<Integer> iterator = source.toIterable().iterator();
    assertTrue(source.isSubscribed());
    TestSubscription subscription = new TestSubscription();
    source.onSubscribe(subscription);
    source.onNext(1, 2);
    assertThat("hasNext timed out.", iterator.hasNext(-1, MILLISECONDS), is(true));
    assertThat("Unexpected item found.", iterator.next(-1, MILLISECONDS), is(1));
    assertThat("Unexpected item found.", iterator.next(-1, MILLISECONDS), is(2));
    expected.expect(instanceOf(TimeoutException.class));
    try {
        iterator.next(10, MILLISECONDS);
    } catch (TimeoutException e) {
        assertThat("Unexpected item found.", iterator.hasNext(-1, MILLISECONDS), is(false));
        assertTrue(subscription.isCancelled());
        throw e;
    }
}
 

/**
 * Waits for the end of the synchronization and updates last seen config
 * file.
 *
 * @param peerId the id of the peer.
 * @throws TimeoutException in case of timeout.
 * @throws IOException in case of IOException.
 * @throws InterruptedException if it's interrupted.
 */
void waitForSyncEnd(String peerId)
    throws TimeoutException, IOException, InterruptedException {
  MessageTuple tuple = filter.getExpectedMessage(MessageType.SYNC_END,
                                                 peerId,
                                                 getSyncTimeoutMs());
  ClusterConfiguration cnf
    = ClusterConfiguration.fromProto(tuple.getMessage().getConfig(),
                                     this.serverId);
  LOG.debug("Got SYNC_END {} from {}", cnf, peerId);
  this.persistence.setLastSeenConfig(cnf);
  if (persistence.isInStateTransfer()) {
    persistence.endStateTransfer();
  }
  // If the synchronization is performed by truncation, then it's possible
  // the content of cluster_config has been truncated in log, then we'll
  // delete these invalid cluster_config files.
  persistence.cleanupClusterConfigFiles();
}
 

@Test
void activateAndDeactivateWithAutoAck() throws IOException, TimeoutException {
  sut = new ConsumerHolder(eventConsumerMock, "queue", true, PREFETCH_COUNT,
      consumerChannelFactoryMock, declarationsListMock, declarerRepositoryMock);
  Assertions.assertEquals("queue", sut.getQueueName());
  Assertions.assertTrue(sut.isAutoAck());
  when(consumerChannelFactoryMock.createChannel()).thenReturn(channelMock);
  sut.activate();
  verify(channelMock).addRecoveryListener(sut);
  verify(channelMock).basicConsume(eq("queue"), eq(true), isA(DeliverCallback.class),
      isA(ConsumerShutdownSignalCallback.class));
  verify(declarerRepositoryMock).declare(channelMock, declarationsListMock);
  verify(channelMock, never()).close();
  verify(channelMock).basicQos(PREFETCH_COUNT);

  sut.deactivate();
  verify(channelMock).close();
}
 

public AttachmentUploadAttributes getAttachmentUploadAttributes() throws IOException {
  try {
    WebSocketRequestMessage requestMessage = WebSocketRequestMessage.newBuilder()
                                                                    .setId(new SecureRandom().nextLong())
                                                                    .setVerb("GET")
                                                                    .setPath("/v2/attachments/form/upload")
                                                                    .build();

    Pair<Integer, String> response = websocket.sendRequest(requestMessage).get(10, TimeUnit.SECONDS);

    if (response.first() < 200 || response.first() >= 300) {
      throw new IOException("Non-successful response: " + response.first());
    }

    return JsonUtil.fromJson(response.second(), AttachmentUploadAttributes.class);
  } catch (InterruptedException | ExecutionException | TimeoutException e) {
    throw new IOException(e);
  }
}
 

@Test
void get() throws Exception {
    final PowerTenant a = mockTenant(harmlessZonky(), true);
    final ExecutorService e = Executors.newFixedThreadPool(1);
    final Scheduler s = Scheduler.create();
    try (final Daemon d = spy(new Daemon(a, lifecycle, s))) {
        assertThat(d.getSessionInfo()).isSameAs(a.getSessionInfo());
        doNothing().when(d)
            .submitWithTenant(any(), any(), any(), any(), any(), any());
        doNothing().when(d)
            .submitTenantless(any(), any(), any(), any(), any(), any());
        final Future<ReturnCode> f = e.submit(d::get); // will block
        assertThatThrownBy(() -> f.get(1, TimeUnit.SECONDS)).isInstanceOf(TimeoutException.class);
        lifecycle.resumeToShutdown(); // unblock
        assertThat(f.get()).isEqualTo(ReturnCode.OK); // should now finish
        // call all the jobs and daemons we know about
        verify(d, times(1)).submitTenantless(any(), any(SimplePayload.class), any(), any(), any(), any());
        verify(d, times(8)).submitWithTenant(any(), any(), any(), any(), any(), any());
    } finally {
        e.shutdownNow();
    }
    verify(a).close();
    assertThat(s.isClosed()).isTrue();
}
 

/**
 * Waits until JVoiceXml is expecting input.
 * 
 * @throws InterruptedException
 *             waiting interrupted
 * @throws TimeoutException
 *             input closed while waiting for input
 * @throws JVoiceXMLEvent
 *             error while waiting
 */
public void waitUntilExpectingInput() throws InterruptedException,
        TimeoutException, JVoiceXMLEvent {
    synchronized (monitor) {
        try {
            if (expectingInput) {
                return;
            }
            monitor.wait();
            if (event != null) {
                throw event;
            }
            if (!expectingInput) {
                throw new TimeoutException(
                      "input closed while waiting for expected input");
            }
        } finally {
            expectingInput = false;
        }
    }
}
 

@Test
void metrics_collector_is_invoked_on_basic_get_consumption() throws IOException, TimeoutException {
    MockConnectionFactory mockConnectionFactory = new MockConnectionFactory();
    SimpleMeterRegistry registry = new SimpleMeterRegistry();
    mockConnectionFactory.setMetricsCollector(new MicrometerMetricsCollector(registry));

    try (MockConnection connection = mockConnectionFactory.newConnection();
         Channel channel = connection.createChannel(42)) {
        String queueName = channel.queueDeclare().getQueue();
        channel.basicPublish("", queueName, null, "".getBytes());

        assertThat(registry.get("rabbitmq.consumed").counter().count()).isEqualTo(0);
        channel.basicGet(queueName, true);
        assertThat(registry.get("rabbitmq.consumed").counter().count()).isEqualTo(1);
    }
}
 

@Test
public void testCancelWait() throws Exception {
    List<Runnable> lr = null;
    JexlScript e = JEXL.createScript("wait(10)");
    Callable<Object> c = e.callable(new TestContext());

    ExecutorService executor = Executors.newFixedThreadPool(1);
    try {
        Future<?> future = executor.submit(c);
        Object t = 42;
        try {
            t = future.get(100, TimeUnit.MILLISECONDS);
            Assert.fail("should have timed out");
        } catch (TimeoutException xtimeout) {
            // ok, ignore
            future.cancel(true);
        }
        Assert.assertTrue(future.isCancelled());
        Assert.assertEquals(42, t);
    } finally {
        lr = executor.shutdownNow();
    }
    Assert.assertTrue(lr.isEmpty());
}
 

@When("^you run your tests? :$")
public void executeTest() throws Exception {
    ExecutorService executor = Executors.newSingleThreadExecutor();
    Future<JupiterExecutionListener> future = executor.submit(
        () -> testEngine.executeTestsForClass(compiledClass.getCompiledClass())
    );
    try {
        JupiterExecutionListener listener = future.get(5, TimeUnit.MINUTES);
        assertThat(listener.allTestsPassed())
            .overridingErrorMessage("Tests should be green")
            .isTrue();
    } catch (TimeoutException e) {
        fail("Tests should have finished");
    } finally {
        executor.shutdownNow();
    }
}
 

@VisibleForTesting
void executeCommand(
    String[] commandLine,
    int timeout,
    String address,
    String workspaceId,
    Set<Integer> successResponseCodes)
    throws TimeoutException, IOException, InterruptedException {
  final ListLineConsumer outputConsumer = new ListLineConsumer();
  Process process = ProcessUtil.executeAndWait(commandLine, timeout, SECONDS, outputConsumer);

  if (!successResponseCodes.contains(process.exitValue())) {
    LOG.error(
        "Error occurred during backup/restore of workspace '{}' on node '{}' : {}",
        workspaceId,
        address,
        outputConsumer.getText());
    throw new IOException("Synchronization process failed. Exit code " + process.exitValue());
  }
}
 

@Override
public T get(long timeoutMs, long sleepingIntervalMs) throws TimeoutException {
	long waitingSince = U.time();

	while (!isDone()) {
		if (U.timedOut(waitingSince, timeoutMs)) {
			throw new TimeoutException();
		}

		U.sleep(sleepingIntervalMs);
	}

	if (getError() != null) {
		throw U.rte("Cannot get the result, there was an error!", error);
	}

	return result;
}
 

@Test
public void negativeDeliveryAck() throws IOException, TimeoutException, InterruptedException {
    //setup
    long deliveryTag = 1L;
    String exchange = "someExchange";
    String routingKey = "someRoutingKey";
    AMQP.BasicProperties properties = new AMQP.BasicProperties();
    byte[] body = new byte[0];
    Delivery delivery = new Delivery(new Envelope(deliveryTag, false, exchange, routingKey), properties, body);

    //test
    differPublisher.negativeDeliveryAck(delivery);

    //assert
    Mockito.verify(channel).basicNack(eq(deliveryTag), eq(false), eq(false));
    Mockito.verify(channel).basicPublish(
            eq(EDDI_EXCHANGE), eq(MESSAGE_CREATED_EDDI_FAILED_ROUTING_KEY), eq(null), eq(body));
    Mockito.verify(channel).waitForConfirmsOrDie(eq(TIMEOUT_CONFIRMS_IN_MILLIS));
}
 

private ReplyMsg doWriteAndSync(final Channel channel, final AskMsg request, final long timeout, final WriteFuture<BaseMsg> writeFuture) throws Exception {

        channel.writeAndFlush(request).addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                writeFuture.setWriteResult(future.isSuccess());
                writeFuture.setCause(future.cause());
                //失败移除
                if (!writeFuture.isWriteSuccess()) {
                    SyncWriteMap.syncKey.remove(writeFuture.requestId());
                }
            }
        });

        ReplyMsg response = (ReplyMsg)writeFuture.get(timeout, TimeUnit.MILLISECONDS);
        if (response == null) {
            if (writeFuture.isTimeout()) {
                throw new TimeoutException();
            } else {
                // write exception
                throw new Exception(writeFuture.cause());
            }
        }
        return response;
    }
 

@Test(expected=IllegalStateException.class)
public void testThrowOnUnsubscribeWhenClosed() throws IOException, InterruptedException, TimeoutException {
    try (NatsTestServer ts = new NatsTestServer(false);
                Connection nc = Nats.connect(ts.getURI())) {
        Dispatcher d = nc.createDispatcher((msg) -> {});
        d.subscribe("foo");
        nc.closeDispatcher(d);
        d.unsubscribe("foo");
        assertFalse(true);
    }
}
 

private void createFile(int fsIdx, Path path, int numBlocks,
    short replicateFactor)
    throws IOException, TimeoutException, InterruptedException {
  final int seed = 0;
  final DistributedFileSystem fs = cluster.getFileSystem(fsIdx);
  DFSTestUtil.createFile(fs, path, BLOCK_SIZE * numBlocks,
      replicateFactor, seed);
  DFSTestUtil.waitReplication(fs, path, replicateFactor);
}
 

protected BackgroundException wrap(final T failure, final String title, final StringBuilder buffer) {
    if(buffer.toString().isEmpty()) {
        log.warn(String.format("No message for failure %s", failure));
        this.append(buffer, LocaleFactory.localizedString("Unknown"));
    }
    for(Throwable cause : ExceptionUtils.getThrowableList(failure)) {
        if(cause instanceof InterruptedIOException) {
            // Handling socket timeouts
            return new ConnectionTimeoutException(buffer.toString(), failure);
        }
        if(cause instanceof TimeoutException) {
            //
            return new ConnectionTimeoutException(buffer.toString(), failure);
        }
        if(cause instanceof SocketException) {
            return new DefaultSocketExceptionMappingService().map((SocketException) cause);
        }
        if(cause instanceof EOFException) {
            return new ConnectionRefusedException(buffer.toString(), failure);
        }
        if(cause instanceof UnknownHostException) {
            return new ResolveFailedException(buffer.toString(), failure);
        }
        if(cause instanceof NoHttpResponseException) {
            return new ConnectionRefusedException(buffer.toString(), failure);
        }
        if(cause instanceof ConnectionClosedException) {
            return new ConnectionRefusedException(buffer.toString(), failure);
        }
        if(cause instanceof InterruptedException) {
            return new ConnectionCanceledException(buffer.toString(), failure);
        }
    }
    if(failure instanceof RuntimeException) {
        return new ConnectionCanceledException(title, buffer.toString(), failure);
    }
    return new BackgroundException(title, buffer.toString(), failure);
}
 

public static void connectToHazelcast() {
    String topic = getRealTopic();
    if (client != null) {
        client.getLifecycleService().removeLifecycleListener(listenerId);
        client.shutdown();
    }
    String[] address = new String[cluster.size()];
    for (int i=0; i < cluster.size(); i++) {
        address[i] = cluster.get(i);
    }
    ClientConnectionStrategyConfig connectionStrategy = new ClientConnectionStrategyConfig();
    connectionStrategy.setReconnectMode(ClientConnectionStrategyConfig.ReconnectMode.ASYNC);
    ConnectionRetryConfig retry = new ConnectionRetryConfig();
    retry.setClusterConnectTimeoutMillis(MAX_CLUSTER_WAIT);
    connectionStrategy.setConnectionRetryConfig(retry);
    ClientConfig config = new ClientConfig();
    config.getNetworkConfig().addAddress(address);
    config.setConnectionStrategyConfig(connectionStrategy);
    client = HazelcastClient.newHazelcastClient(config);
    client.getCluster().addMembershipListener(new ClusterListener());
    listenerId = client.getLifecycleService().addLifecycleListener(new TopicLifecycleListener(topic));
    if (!isServiceMonitor) {
        try {
            // recover the topic
            PostOffice.getInstance().request(MANAGER, 10000, new Kv(TYPE, TopicManager.CREATE_TOPIC));
        } catch (IOException | TimeoutException | AppException e) {
            log.error("Unable to create topic {} - {}", topic, e.getMessage());
        }
    }
    log.info("Connected to hazelcast cluster and listening to {} ", topic);
}
 

private static IResponse dispatchHandler(IRequest request, IResponse response) {
    try {
        FutureTask<IResponse> futureTask = (FutureTask<IResponse>) ThreadUtil.execAsync(new RequestTask(request, response));
        if (RobotUtil.isDevMode()) {
            response = futureTask.get();
        } else {
            response = futureTask.get(REQUEST_TIME_OUT, TimeUnit.MILLISECONDS);
        }
        if (response.getStatus() == HttpStatus.HTTP_OK) {
            // 不是业务请求且不是完成请求的都需要添加到重发队列
            if (!RobotUtil.isBusinessRequest(request)
                    && !RobotUtil.isOrderStateRequest(request)
                    && request.isNeedRepeatSend()) {
                // 将Response对象放入重发队列，确保消息发送到车辆
                RobotUtil.getRepeatSend().add(response);
            }
            //是同一单元的请求响应且需要发送的响应
            if (response.isResponseTo(request) && response.isNeedSend()) {
                // 正确的响应才发送到车辆或设备
                TelegramSendKit.duang().key(response.getDeviceId()).response(response).send();
            }
        } else {
            createResponseException(request, response, response.getException());
        }
    } catch (TimeoutException te) {
        createResponseException(request, response, te);
    } catch (Exception e) {
        createResponseException(request, response, e);
    }
    return response;
}
 

public String invokeAPI(String host, String path, String method, List<Pair> queryParams, Object body, Map<String, String> headerParams, Map<String, String> formParams, String contentType, String[] authNames) throws ApiException, InterruptedException, ExecutionException, TimeoutException {
  try {
    RequestFuture<String> future = RequestFuture.newFuture();
    Request request = createRequest(host, path, method, queryParams, body, headerParams, formParams, contentType, authNames, future, future);
    if(request != null) {
       mRequestQueue.add(request);
       return future.get(connectionTimeout, TimeUnit.SECONDS);
    } else {
      return "no data";
    }
  } catch (UnsupportedEncodingException ex) {
    throw new ApiException(0, "UnsupportedEncodingException");
  }
}
 

@Test
public void serviceThatCantBeBoundTo() {
  Intent intent = new Intent(getApplicationContext(), ServiceThatCantBeBoundTo.class);
  try {
    serviceRule.startService(intent);
    fail("TimeoutException was not thrown");
  } catch (TimeoutException e) {
    // expected
  }
}
 

@Bean("mqConnectionReceive")
public Connection mqConnectionReceive(@Autowired MQConfigBean mqConfigBean) throws IOException, TimeoutException {
    ConnectionFactory factory = new ConnectionFactory();
    //用户名
    factory.setUsername(username);
    //密码
    factory.setPassword(password);
    //虚拟主机路径（相当于数据库名）
    factory.setVirtualHost(virtualHost);
    //返回连接
    return factory.newConnection(mqConfigBean.getAddressList());
}
 

@AfterClass
public static void teardown() throws InterruptedException, ExecutionException, TimeoutException {
	if (rpcService != null) {
		RpcUtils.terminateRpcService(rpcService, TIMEOUT);
		rpcService = null;
	}
}
 

@Test
void createConsumerNameIsNotSetAndBridgeIdIsSet(VertxTestContext context) throws InterruptedException, ExecutionException, TimeoutException {
    JsonObject json = new JsonObject();
    String[] consumerInstanceId = {""};

    CompletableFuture<Boolean> create = new CompletableFuture<>();
    consumerService()
        .createConsumerRequest(groupId, json)
        .as(BodyCodec.jsonObject())
        .sendJsonObject(json, ar -> {
            context.verify(() -> {
                assertThat(ar.succeeded(), is(true));
                HttpResponse<JsonObject> response = ar.result();
                assertThat(response.statusCode(), is(HttpResponseStatus.OK.code()));
                JsonObject bridgeResponse = response.body();
                consumerInstanceId[0] = bridgeResponse.getString("instance_id");
                assertThat(consumerInstanceId[0], startsWith("my-bridge-"));
            });
            create.complete(true);
        });

    create.get(TEST_TIMEOUT, TimeUnit.SECONDS);
    consumerService()
        .deleteConsumer(context, groupId, consumerInstanceId[0]);
    context.completeNow();

}
 

@Test
public void queryExchangeRate_shouldBeGetMethod()
        throws InterruptedException, TimeoutException {

    exchangeApi.queryExchangeRate("XMR", "USD", mockExchangeCallback);

    RecordedRequest request = mockWebServer.takeRequest();
    assertEquals("GET", request.getMethod());
}
 

@Test
public void onceFiresExactlyOnce()
    throws DatabaseException, TestFailure, ExecutionException, TimeoutException,
        InterruptedException {
  DatabaseReference ref = IntegrationTestHelpers.getRandomNode();

  final AtomicBoolean called = new AtomicBoolean(false);
  ref.addListenerForSingleValueEvent(
      new ValueEventListener() {
        @Override
        public void onDataChange(DataSnapshot snapshot) {
          assertTrue(called.compareAndSet(false, true));
        }

        @Override
        public void onCancelled(DatabaseError error) {
          fail("Should not be cancelled");
        }
      });

  ZombieVerifier.verifyRepoZombies(ref);

  ref.setValue(42);
  ref.setValue(84);
  new WriteFuture(ref, null).timedGet();
  ZombieVerifier.verifyRepoZombies(ref);
}
 

@Test
public void testStatsRequest() throws InterruptedException, ExecutionException, TimeoutException {
    System.out.println("Sending the stats request");
    List<? extends MaestroNote> replies = maestro
            .statsRequest()
            .get(10, TimeUnit.SECONDS);

    assertEquals("Unexpected reply size", 2, replies.size());

    MaestroNote note = replies.get(0);
    assertEquals(note.getNoteType(), MaestroNoteType.MAESTRO_TYPE_RESPONSE);
    assertEquals(note.getMaestroCommand(), MaestroCommand.MAESTRO_NOTE_STATS);
}
 

@Override
public boolean delete(final String key) throws TimeoutException, CacheException {
    try {
        return memcachedClient.delete(key).get();
    } catch (InterruptedException | ExecutionException e) {
        throw new CacheException(e);
    }
}