Java 类com.rabbitmq.client.GetResponse 实例源码

/**
 * Helper method to retrieve queue message from rabbitMQ
 *
 * @return result
 * @throws Exception
 */
private static String consumeWithoutCertificate() throws Exception {
    String result = "";
    ConnectionFactory factory = new ConnectionFactory();
    factory.setHost("localhost");
    factory.setPort(5671);
    factory.useSslProtocol();

    Connection conn = factory.newConnection();
    Channel channel = conn.createChannel();

    GetResponse chResponse = channel.basicGet("WithoutClientCertQueue", true);
    if(chResponse != null) {
        byte[] body = chResponse.getBody();
        result = new String(body);
    }
    channel.close();
    conn.close();
    return result;
}

@Test
public void basicGet() throws Exception {
  String exchangeName = "basicGetExchange";
  String queueName = "basicGetQueue";
  String routingKey = "#";

  channel.exchangeDeclare(exchangeName, "direct", true);
  channel.queueDeclare(queueName, true, false, false, null);
  channel.queueBind(queueName, exchangeName, routingKey);

  byte[] messageBodyBytes = "Hello, world!".getBytes();

  channel.basicPublish(exchangeName, routingKey, null, messageBodyBytes);

  GetResponse response = channel.basicGet(queueName, false);
  assertNotNull(response.getBody());

  List<MockSpan> finishedSpans = mockTracer.finishedSpans();
  assertEquals(2, finishedSpans.size());
  checkSpans(finishedSpans);

  assertNull(mockTracer.activeSpan());
}

@Override
public void basicGet(String queue, boolean autoAck, Handler<AsyncResult<JsonObject>> resultHandler) {
  forChannel(resultHandler, (channel) -> {
    GetResponse response = channel.basicGet(queue, autoAck);
    if (response == null) {
      return null;
    } else {
      JsonObject json = new JsonObject();
      populate(json, response.getEnvelope());
      if (includeProperties) {
        put("properties", Utils.toJson(response.getProps()), json);
      }
      put("body", parse(response.getProps(), response.getBody()), json);
      Utils.put("messageCount", response.getMessageCount(), json);
      return json;
    }
  });
}

/**
 * {@inheritDoc}
 *
 * @throws QueueException.EphemeralIsFull
 *             if the ephemeral storage is full
 */
@Override
public IQueueMessage<ID, DATA> take() throws QueueException.EphemeralIsFull {
    try {
        GetResponse msg = getConsumerChannel().basicGet(queueName, true);
        return msg != null ? deserialize(msg.getBody()) : null;
    } catch (Exception e) {
        throw e instanceof QueueException ? (QueueException) e : new QueueException(e);
    }

    // try (Channel channel = createChannel()) {
    // GetResponse msg = channel.basicGet(queueName, true);
    // return msg != null ? deserialize(msg.getBody()) : null;
    // } catch (Exception e) {
    // throw e instanceof QueueException ? (QueueException) e : new
    // QueueException(e);
    // }
}

@Override
public void recover() {
    try {
        componentManager.getHeartbeatMQAccessService().start();
        stateManager = componentManager.getStateManager();

        LOG.info("Start to read messages from MQ ...");
        final Channel channel = componentManager.getHeartbeatMQAccessService().getChannel();
        final String q = componentManager.getHeartbeatMQAccessService().getQueueName();

        while(true) {
            GetResponse response = channel.basicGet(q, false);
            if(response != null) {
                long deliveryTag = response.getEnvelope().getDeliveryTag();
                String message = new String(response.getBody(), "UTF-8");
                LOG.info("Received msg: deliveryTag=" + deliveryTag + ", message=" + message);

                JSONObject result = JSONObject.parseObject(message);
                if(result.containsKey(JSONKeys.TYPE)) {
                    String type = result.getString(JSONKeys.TYPE);
                    switch(type) {
                        case ScheduledConstants.HEARTBEAT_TYPE_TASK_PROGRESS:

                            // {"type":"taskProgress","platform_id":"7fe13e9879314da38bb7abc8b61657bb","tasks":[{"result":{"traceId":"1480402382967","callId":"1","resultCount":"1423","message":"SUCCESS","status":5},"jobId":1,"taskType":1,"taskId":1,"seqNo":1,"status":"SUCCEEDED"}]}
                            // add flag 'needRecovering' to a tasks response
                            result.put(ScheduledConstants.NEED_RECOVERING, ScheduledConstants.YES);
                            needRecoveredTaskQueue.add(result);
                            channel.basicAck(deliveryTag, false);
                            break;
                    }
                } else {
                    channel.basicAck(deliveryTag, false);
                }
            } else {
                break;
            }
        }
        LOG.info("Complete to read MQ messages: q=" + q);

        // update Redis statuses
        recoverRedisStates();

        // recover job/task statuses
        processPendingTaskResponses();
    } catch (Exception e) {
        LOG.error("Recovery failure:", e);
        Throwables.propagate(e);
    } finally {
        componentManager.getHeartbeatMQAccessService().stop();
    }
}

public List<String> popAllMessages() throws IOException, InterruptedException {
    List<String> messages = new ArrayList<>();
    GetResponse response;

    while ((response = channel.basicGet(routeKey, true)) != null) {
        messages.add(new String(response.getBody()));
    }
    return messages;
}

/**
 * Helper method to retrieve queue message from rabbitMQ
 *
 * @return result
 * @throws Exception
 */
private static String consumeWithoutCertificate() throws Exception {
    String result = "";

    String basePath = TestConfigurationProvider.getResourceLocation() + "/artifacts/ESB/messageStore/rabbitMQ/SSL/";

    String truststoreLocation = basePath + "rabbitMQ/certs/client/rabbitstore";
    String keystoreLocation = basePath + "rabbitMQ/certs/client/keycert.p12";

    char[] keyPassphrase = "MySecretPassword".toCharArray();
    KeyStore ks = KeyStore.getInstance("PKCS12");
    ks.load(new FileInputStream(keystoreLocation), keyPassphrase);

    KeyManagerFactory kmf = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
    kmf.init(ks, keyPassphrase);

    char[] trustPassphrase = "rabbitstore".toCharArray();
    KeyStore tks = KeyStore.getInstance("JKS");
    tks.load(new FileInputStream(truststoreLocation), trustPassphrase);

    TrustManagerFactory tmf = TrustManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
    tmf.init(tks);

    SSLContext c = SSLContext.getInstance("SSL");
    c.init(kmf.getKeyManagers(), tmf.getTrustManagers(), null);

    ConnectionFactory factory = new ConnectionFactory();
    factory.setHost("localhost");
    factory.setPort(5671);
    factory.useSslProtocol(c);

    Connection conn = factory.newConnection();
    Channel channel = conn.createChannel();

    GetResponse chResponse = channel.basicGet("WithClientCertQueue", true);
    if(chResponse != null) {
        byte[] body = chResponse.getBody();
        result = new String(body);
    }
    channel.close();
    conn.close();
    return result;
}

@Override
public T consumeNext() {
    try {
        GetResponse response = channel.basicGet(name, true);
        return MAPPER.readValue(response.getBody(), type);
    } catch (IOException e) {
        throw new MessageQueueException("Unable to consume.", e);
    }
}

public GetResponse getMessage() throws IOException {
    GetResponse response = null;
    try {
        response = channel.basicGet(queueName, false);
    } catch (IOException e) {
        logger.error("Error getting message from queue '" + queueName + "'.");
        throw e;
    }

    return response;
}

@Override
public GetResponse basicGet(String queue, boolean autoAck) throws IOException {
  GetResponse response = channel.basicGet(queue, autoAck);
  TracingUtils.buildAndFinishChildSpan(response.getProps(), tracer);
  return response;
}

@Test
public void prePutHappyCase() throws Exception {

    Map<String, String> kvs = configureHBase(primaryTableNameString, secondaryIdxTableNameString, "e", "eg", "a", amq_default_address, primaryTableNameString, "true", "");
    setupHBase(kvs);

    //simulate population of secondary index as a result of the above
    Put idxPut = new Put("EFG1".getBytes());
    idxPut.addColumn("a".getBytes(), "EFB1".getBytes(), "".getBytes());
    secondaryIdxTable.put(idxPut);

    // Add a column to the primary table, which should trigger a data ripple to the downstream table
    Put tablePut = new Put("EFG1".getBytes());
    tablePut.addColumn("e".getBytes(), "some_key".getBytes(), "some_value".getBytes());
    primaryTable.put(tablePut);

    //check that values made it to the queue
    com.rabbitmq.client.ConnectionFactory factory = new com.rabbitmq.client.ConnectionFactory();
    factory.setUri(amq_default_address);
    com.rabbitmq.client.Connection conn = factory.newConnection();
    com.rabbitmq.client.Channel channel = conn.createChannel();
    System.out.println(String.format("Test: connecting to %s", primaryTableNameString));

    while (true) {
        GetResponse response = channel.basicGet(primaryTableNameString, false);
        if (response == null)//busy-wait until the message has made it through the MQ
        {
            continue;
        }
        String routingKey = response.getEnvelope().getRoutingKey();
        Assert.assertEquals("Routing key should be rowkey", "genome", routingKey);

        String contentType = response.getProps().getContentType();
        Assert.assertEquals("Content type should be preserved", "application/json", contentType);

        Map<String, Object> headers = response.getProps().getHeaders();
        Assert.assertEquals("An action should be set on the message", "put", headers.get("action").toString());

        byte[] body = response.getBody();

        JSONObject jo = new JSONObject(new String(body));
        String column_family = (String) jo.get("column_family");
        Assert.assertEquals("Column family should be preserved in the message body", "eg", column_family);

        String column_value = (String) jo.get("column_value");
        Assert.assertEquals("Column value should be preserved in the message body", "some_value", column_value);

        long deliveryTag = response.getEnvelope().getDeliveryTag();
        channel.basicAck(deliveryTag, false);
        break;
    }
}

@Test
public void verifyValueNotSentByDefault() throws Exception {

    Map<String, String> kvs = configureHBase(primaryTableNameString, secondaryIdxTableNameString, "e", "eg", "a", amq_default_address, primaryTableNameString, "false", "");
    setupHBase(kvs);

    //simulate population of secondary index as a result of the above
    Put idxPut = new Put("EFG1".getBytes());
    idxPut.addColumn("a".getBytes(), "EFB1".getBytes(), "".getBytes());
    secondaryIdxTable.put(idxPut);

    // Add a column to the primary table, which should trigger a data ripple to the downstream table
    Put tablePut = new Put("EFG1".getBytes());
    tablePut.addColumn("e".getBytes(), "some_key".getBytes(), "some_value".getBytes());
    primaryTable.put(tablePut);

    //check that values made it to the queue
    com.rabbitmq.client.ConnectionFactory factory = new com.rabbitmq.client.ConnectionFactory();
    factory.setUri(amq_default_address);
    com.rabbitmq.client.Connection conn = factory.newConnection();
    com.rabbitmq.client.Channel channel = conn.createChannel();
    System.out.println(String.format("Test: connecting to %s", primaryTableNameString));

    while (true) {
        GetResponse response = channel.basicGet(primaryTableNameString, false);
        if (response == null)//busy-wait until the message has made it through the MQ
        {
            continue;
        }
        String routingKey = response.getEnvelope().getRoutingKey();
        Assert.assertEquals("Routing key should be rowkey", "genome", routingKey);

        String contentType = response.getProps().getContentType();
        Assert.assertEquals("Content type should be preserved", "application/json", contentType);

        Map<String, Object> headers = response.getProps().getHeaders();
        Assert.assertEquals("An action should be set on the message", "put", headers.get("action").toString());

        byte[] body = response.getBody();

        JSONObject jo = new JSONObject(new String(body));
        String column_family = (String) jo.get("column_family");
        Assert.assertEquals("Column family should be preserved in the message body", "eg", column_family);

        String column_value = (String) jo.get("column_value");
        Assert.assertEquals("Column value is not sent by default", "", column_value);

        long deliveryTag = response.getEnvelope().getDeliveryTag();
        channel.basicAck(deliveryTag, false);
        break;
    }
}

@Test
public void filterOnQualifiers() throws Exception {

    Map<String, String> kvs = configureHBase(primaryTableNameString, secondaryIdxTableNameString, "e", "eg", "a", amq_default_address, primaryTableNameString, "false", "one_key|some_key");
    setupHBase(kvs);

    //simulate population of secondary index as a result of the above
    Put idxPut = new Put("EFG1".getBytes());
    idxPut.addColumn("a".getBytes(), "EFB1".getBytes(), "".getBytes());
    secondaryIdxTable.put(idxPut);

    // Add a column to the primary table, which should trigger a data ripple to the downstream table
    Put tablePut = new Put("EFG1".getBytes());
    tablePut.addColumn("e".getBytes(), "some_key".getBytes(), "some_value".getBytes());
    primaryTable.put(tablePut);

    //check that values made it to the queue
    com.rabbitmq.client.ConnectionFactory factory = new com.rabbitmq.client.ConnectionFactory();
    factory.setUri(amq_default_address);
    com.rabbitmq.client.Connection conn = factory.newConnection();
    com.rabbitmq.client.Channel channel = conn.createChannel();
    System.out.println(String.format("Test: connecting to %s", primaryTableNameString));

    while (true) {
        GetResponse response = channel.basicGet(primaryTableNameString, false);
        if (response == null)//busy-wait until the message has made it through the MQ
        {
            continue;
        }
        String routingKey = response.getEnvelope().getRoutingKey();
        Assert.assertEquals("Routing key should be rowkey", "genome", routingKey);

        String contentType = response.getProps().getContentType();
        Assert.assertEquals("Content type should be preserved", "application/json", contentType);

        Map<String, Object> headers = response.getProps().getHeaders();
        Assert.assertEquals("An action should be set on the message", "put", headers.get("action").toString());

        byte[] body = response.getBody();

        JSONObject jo = new JSONObject(new String(body));
        String column_family = (String) jo.get("column_family");
        Assert.assertEquals("Column family should be preserved in the message body", "eg", column_family);

        String column_value = (String) jo.get("column_value");
        Assert.assertEquals("Column value is not sent by default", "", column_value);

        long deliveryTag = response.getEnvelope().getDeliveryTag();
        channel.basicAck(deliveryTag, false);
        break;
    }
}

@Test
public void onlyPutWhenInQualifierFilter() throws Exception {

    Map<String, String> kvs = configureHBase(primaryTableNameString, secondaryIdxTableNameString, "e", "eg", "a", amq_default_address, primaryTableNameString, "false", "some_key");
    setupHBase(kvs);

    //simulate population of secondary index as a result of the above
    Put idxPut = new Put("EFG1".getBytes());
    idxPut.addColumn("a".getBytes(), "EFB1".getBytes(), "".getBytes());
    secondaryIdxTable.put(idxPut);

    // Add a column to the primary table, which should trigger a data ripple to the downstream table
    Put tablePut = new Put("EFG1".getBytes());
    tablePut.addColumn("e".getBytes(), "some_key".getBytes(), "some_value".getBytes());
    tablePut.addColumn("e".getBytes(), "other_key".getBytes(), "some_value".getBytes());
    tablePut.addColumn("e".getBytes(), "third_key".getBytes(), "some_value".getBytes());
    primaryTable.put(tablePut);

    //check that values made it to the queue
    com.rabbitmq.client.ConnectionFactory factory = new com.rabbitmq.client.ConnectionFactory();
    factory.setUri(amq_default_address);
    com.rabbitmq.client.Connection conn = factory.newConnection();
    com.rabbitmq.client.Channel channel = conn.createChannel();
    System.out.println(String.format("Test: connecting to %s", primaryTableNameString));

    while (true) {
 int numMessages = channel.queueDeclarePassive(primaryTableNameString).getMessageCount();
        GetResponse response = channel.basicGet(primaryTableNameString, false);
        if (response == null || numMessages == 0)//busy-wait until the message has made it through the MQ
        {
            continue;
        }
 Assert.assertEquals("There should be only a single key in the queue", 1, numMessages);
        String routingKey = response.getEnvelope().getRoutingKey();
        Assert.assertEquals("Routing key should be rowkey", "genome", routingKey);

        String contentType = response.getProps().getContentType();
        Assert.assertEquals("Content type should be preserved", "application/json", contentType);

        Map<String, Object> headers = response.getProps().getHeaders();
        Assert.assertEquals("An action should be set on the message", "put", headers.get("action").toString());

        byte[] body = response.getBody();

        JSONObject jo = new JSONObject(new String(body));
        String column_family = (String) jo.get("column_family");
        Assert.assertEquals("Column family should be preserved in the message body", "eg", column_family);

        String column_value = (String) jo.get("column_value");
        Assert.assertEquals("Column value is not sent by default", "", column_value);

        long deliveryTag = response.getEnvelope().getDeliveryTag();
        channel.basicAck(deliveryTag, false);
        break;
    }
}

@Test
public void preDeleteHappyCase() throws Exception {

    Map<String, String> kvs = configureHBase(primaryTableNameString, secondaryIdxTableNameString, "e", "eg", "a", amq_default_address, primaryTableNameString, "true", "");
    setupHBase(kvs);
    com.rabbitmq.client.ConnectionFactory factory = new com.rabbitmq.client.ConnectionFactory();
    factory.setUri(amq_default_address);
    com.rabbitmq.client.Connection conn = factory.newConnection();
    com.rabbitmq.client.Channel channel = conn.createChannel();


    //simulate population of secondary index for a put on the downstreamTable
    Put idxPut = new Put("EFG1".getBytes());
    idxPut.addColumn("a".getBytes(), "EFB1".getBytes(), "".getBytes());
    secondaryIdxTable.put(idxPut);

    //simulate a data rippling performed by the data_rippler service consuming off of the rabbitmq queue
    Put tablePut = new Put("EFB1".getBytes());
    tablePut.addColumn("eg".getBytes(), "some_key".getBytes(), "some_value".getBytes());
    downstreamTable.put(tablePut);

    // since we made no active put to the queue from the prePut, we need to declare it explicitly here
    channel.queueDeclare(primaryTableNameString, true, false, false, null);

    // finished with the setup, we now issue a delete which should be caught by the rabbitmq
    Delete d = new Delete("EFG1".getBytes());
    primaryTable.delete(d);

    //check that values made it to the queue
    while (true) {
        GetResponse response = channel.basicGet(primaryTableNameString, false);
        if (response == null)//busy-wait until the message has made it through the MQ
        {
            continue;
        }
        String routingKey = response.getEnvelope().getRoutingKey();
        Assert.assertEquals("Routing key should be rowkey", "genome", routingKey);

        String contentType = response.getProps().getContentType();
        Assert.assertEquals("Content type should be preserved", "application/json", contentType);

        Map<String, Object> headers = response.getProps().getHeaders();
        Assert.assertEquals("An action should be set on the message", "delete", headers.get("action").toString());

        byte[] body = response.getBody();
        JSONObject jo = new JSONObject(new String(body));

        String column_qualifier = (String) jo.get("column_qualifier");
        Assert.assertEquals("Column qualifier should be empty, signalling a row delete", "", column_qualifier);

        long deliveryTag = response.getEnvelope().getDeliveryTag();
        channel.basicAck(deliveryTag, false);
        break;
    }
}

@Test
public void discernNewPutFromUpdate() throws Exception {
    Map<String, String> kvs = configureHBase(primaryTableNameString, secondaryIdxTableNameString, "e", "eg", "a", amq_default_address, primaryTableNameString, "true", "");
    setupHBase(kvs);

    //simulate population of secondary index as a result of the above
    Put idxPut = new Put("EFG1".getBytes());
    idxPut.addColumn("a".getBytes(), "EFB1".getBytes(), "".getBytes());
    secondaryIdxTable.put(idxPut);

    /*
        The first put should be registered as a "put" action, while the second should be registered as an "update"
        action, thereby signalling different action to be taken by the consumers
     */

    Put tablePut = new Put("EFG1".getBytes());
    tablePut.addColumn("e".getBytes(), "some_key".getBytes(), "some_value".getBytes());
    primaryTable.put(tablePut);

    tablePut = new Put("EFG1".getBytes());
    tablePut.addColumn("e".getBytes(), "some_other_key".getBytes(), "some_value".getBytes());
    primaryTable.put(tablePut);

    //check that values made it to the queue
    com.rabbitmq.client.ConnectionFactory factory = new com.rabbitmq.client.ConnectionFactory();
    factory.setUri(amq_default_address);
    com.rabbitmq.client.Connection conn = factory.newConnection();
    com.rabbitmq.client.Channel channel = conn.createChannel();
    int msgs_to_consume = 2;
    while (msgs_to_consume > 0) {
        System.out.println(String.format("Messages to get: %s", msgs_to_consume));
        GetResponse response = channel.basicGet(primaryTableNameString, false);
        if (response == null)//busy-wait until the message has made it through the MQ
        {
            continue;
        }
        String routingKey = response.getEnvelope().getRoutingKey();
        Assert.assertEquals("Routing key should be rowkey", "genome", routingKey);

        String contentType = response.getProps().getContentType();
        Assert.assertEquals("Content type should be preserved", "application/json", contentType);

        Map<String, Object> headers = response.getProps().getHeaders();
        byte[] body = response.getBody();

        JSONObject jo = new JSONObject(new String(body));
        String column_family = (String) jo.get("column_family");
        Assert.assertEquals("Column family should be preserved in the message body", "eg", column_family);

        String column_value = (String) jo.get("column_qualifier");

        if(headers.get("action").toString().equals("update")){
            Assert.assertEquals("Column value should be preserved in the message body", "some_other_key", column_value);
        }
        else {
            Assert.assertEquals("Column value should be preserved in the message body", "some_key", column_value);
        }

        long deliveryTag = response.getEnvelope().getDeliveryTag();
        channel.basicAck(deliveryTag, false);
        msgs_to_consume--;
    }
}

@Override
public GetResponse basicGet(String queue, boolean autoAck) throws IOException {
    return delegate.basicGet(queue,autoAck);
}

/**
 * Retrieve a message from a queue using {@link com.rabbitmq.client.AMQP.Basic.Get}
 * @see com.rabbitmq.client.AMQP.Basic.Get
 * @see com.rabbitmq.client.AMQP.Basic.GetOk
 * @see com.rabbitmq.client.AMQP.Basic.GetEmpty
 * @param queue the name of the queue
 * @param autoAck true if the server should consider messages
 * acknowledged once delivered; false if the server should expect
 * explicit acknowledgements
 * @return a {@link GetResponse} containing the retrieved message data
 * @throws java.io.IOException if an error is encountered
 */
GetResponse basicGet(String queue, boolean autoAck) throws IOException;

/**
 * Retrieve a message from a queue using {@link com.rabbitmq.client.AMQP.Basic.Get}
 * @see com.rabbitmq.client.AMQP.Basic.Get
 * @see com.rabbitmq.client.AMQP.Basic.GetOk
 * @see com.rabbitmq.client.AMQP.Basic.GetEmpty
 * @param queue the name of the queue
 * @param autoAck true if the server should consider messages
 * acknowledged once delivered; false if the server should expect
 * explicit acknowledgements
 * @return a {@link GetResponse} containing the retrieved message data
 * @throws java.io.IOException if an error is encountered
 */
GetResponse basicGet(String queue, boolean autoAck) throws IOException;