在 Apache Edgent 中加入不同类型的流答案

【问题标题】：Joining streams of different types in Apache Edgent在 Apache Edgent 中加入不同类型的流
【发布时间】：2017-02-02 23:50:57
【问题描述】：

我有 3 个流：

TStream<Double> tempReadings=topology.poll(tempSensor, 10, TimeUnit.SECONDS);
TStream<Double> co2Readings=topology.poll(co2Sensor, 10, TimeUnit.SECONDS);
TStream<Boolean> stationaryReadings=topology.poll(stationarySensor, 10, TimeUnit.SECONDS);

我目前从 3 个 JSON 对象创建 3 个单独的设备事件：

TStream<JsonObject> tempJson=tempReadings.map(tuple->{
    JsonObject json=new JsonObject();
    json.addProperty("Temperature", tuple);
    return json;
});
TStream<JsonObject> co2Json=co2Readings.map(tuple->{
    JsonObject json=new JsonObject();
    json.addProperty("C02Level", tuple);
    return json;
});
TStream<JsonObject> sensoryJson=stationaryReadings.map(tuple->{
    JsonObject json=new JsonObject();
    json.addProperty("isStationary", tuple);
    return json;
});

我想通过将这些流连接在一起并创建 1 个具有三个属性（Temperature、C02Level 和 isStationary）的 JSON 对象来创建单个事件。

【问题讨论】：

您可以union 流，但这只会将一个项目放在另一个项目之后。如果您想一次读取所有 3 个属性，则可以制作一个新传感器，该传感器返回一个包含所有 3 个属性的“读数”对象。
@approxiblue 这是真的，谢谢。我只是想知道是否有更“整洁”的方式来做到这一点，但如果没有，那就没有。如果您希望我将您的答案添加为已接受的答案，请将其写为答案而不是评论。

标签： java json apache-edgent

【解决方案1】：

您可以union 流，但这只会将一个元组一个接一个地放置，并且您需要使用相同类型的流。

如果您想一次读取所有 3 个属性，您可以创建一个返回“读数”对象的传感器：

class Reading {
    Double temperature;
    Double c02Level;
    Boolean isStationary;
}

【讨论】：

【解决方案2】：

在这种情况下，“单轮询结合阅读元组”的方法可能是最好的。

更一般地说，PlumbingStreams.barrier() 可用于合并多个流的对应元组。比如：

TStream<JsonObject> combinedReadings =
    PlumbingStreams.barrier(Arrays.asList(tempJson,co2Json,sensoryJson))
    .map(list -> combineTuples(list));


static JsonObject combineTuples(JsonObject list...) {
  JsonObject jo = new JsonObject();
  for (JsonObject j : list) {
    for (Entry<String,JsonElement> e : j.entrySet()) {
      jo.addProperty(e.getKey(), e.getValue());
    }
  }
  return jo;
}

【讨论】：