public static Read read() {
    return new AutoValue_RabbitMqIO_Read.Builder()
        .setQueueDeclare(false)
        .setExchangeDeclare(false)
        .setMaxReadTime(null)
        .setMaxNumRecords(Long.MAX_VALUE)
        .setUseCorrelationId(false)
        .build();
  }
@Test
  public void testReadDeclaredTopicExchange() throws Exception {
    final int numRecords = 10;
    RabbitMqIO.Read read =
        RabbitMqIO.read().withExchange("DeclaredTopicExchange", "topic", "user.create.#");

    final Supplier<String> publishRoutingKeyGen =
        new Supplier<String>() {
          private AtomicInteger counter = new AtomicInteger(0);

          @Override
          public String get() {
            int count = counter.getAndIncrement();
            if (count % 2 == 0) {
              return "user.create." + count;
            }
            return "user.delete." + count;
          }
        };

    ExchangeTestPlan plan =
        new ExchangeTestPlan(read, numRecords / 2, numRecords) {
          @Override
          public Supplier<String> publishRoutingKeyGen() {
            return publishRoutingKeyGen;
          }

          @Override
          public List<String> expectedResults() {
            return IntStream.range(0, numRecords)
                .filter(i -> i % 2 == 0)
                .mapToObj(RabbitMqTestUtils::generateRecord)
                .map(RabbitMqTestUtils::recordToString)
                .collect(Collectors.toList());
          }
        };

    doExchangeTest(plan);
  }