如何在 PubSub 中接收属性

Question

谁能帮我从 PubSub-Message 接收属性？消息显示正确，但没有属性。我的 Receive-Class 如下所示：

public class ReceiveMessageServlet extends HttpServlet {

@Override
public final void doPost(final HttpServletRequest req,
                         final HttpServletResponse resp)
        throws IOException {
    // Validating unique subscription token before processing the message
    String subscriptionToken = System.getProperty(
            Constants.BASE_PACKAGE + ".subscriptionUniqueToken");
    if (!subscriptionToken.equals(req.getParameter("token"))) {
        resp.setStatus(HttpServletResponse.SC_BAD_REQUEST);
        resp.getWriter().close();
        return;
    }

    ServletInputStream inputStream = req.getInputStream();

    // Parse the JSON message to the POJO model class
    JsonParser parser = JacksonFactory.getDefaultInstance()
            .createJsonParser(inputStream);
    parser.skipToKey("message");
    PubsubMessage message = parser.parseAndClose(PubsubMessage.class);

    // Store the message in the datastore
    Entity messageToStore = new Entity("PubsubMessage");
    messageToStore.setProperty("message",
            new String(message.decodeData(), "UTF-8"));
    messageToStore.setProperty("receipt-time", System.currentTimeMillis());
    DatastoreService datastore =
            DatastoreServiceFactory.getDatastoreService();
    datastore.put(messageToStore);

    // Invalidate the cache
    MemcacheService memcacheService =
            MemcacheServiceFactory.getMemcacheService();
    memcacheService.delete(Constants.MESSAGE_CACHE_KEY);

    // Acknowledge the message by returning a success code
    resp.setStatus(HttpServletResponse.SC_OK);
    resp.getWriter().close();


}

}

我已经用 message.getAttributes() 尝试过，但没有任何成功。

Answer 1

您可能需要检查添加的 PubsubIO 功能，该功能允许 Read 和 Write 转换提供对已添加到 latest release 中的 Cloud Pub/Sub 消息属性的访问。

尝试使用withAttributes 方法。这个，

使源返回一个包含 Pubsub 属性的 PubsubMessage。用户必须提供解析函数来将 PubsubMessage 转换为输出类型。必须通过PCollection.setCoder(Coder) 在输出上注册或设置输出类型 T 的编码器。

希望对你有帮助。