Реактивные конечные точки в Quarkus возвращаются по обмену результаты ⇐ JAVA

[Anonymous]

Я довольно новичок во всем мире Reactive/Mutiny/Vert.x, и у меня есть следующая проблема, которую я не могу понять:
У меня есть две конечные точки отдыха, они оба получают некоторые данные (количество) для определенного устройства (или списка устройств). Единственная разница между двумя конечными точками - это Postgres (TimeScaledB). Теперь, когда я вызываю две конечные точки одновременно, иногда я получаю результаты замены. У меня нет идей, и больше всего я не понимаю, как это может произойти.

Код: Выделить всё

@Path("/api/prefix/foo")
public class DeviceCountResource {

private static final Logger logger = Logger.getLogger(DeviceCountResource.class);

@Inject
DeviceEventService deviceEventService;

@GET
@Produces(MediaType.APPLICATION_JSON)
@Path("{agent_id}/device/status-change-count")
public Uni getDeviceStatusChangeCount(
@PathParam("agent_id") Long agentId,
@QueryParam("device_ids") @Separator(",") List deviceIds
) {
return deviceEventService.getStatusChangeCount(agentId, deviceIds)
.invoke(counters ->
logger.debugf(
"Retrieved counters status change: %s",
counters
));
}

@GET
@Produces(MediaType.APPLICATION_JSON)
@Path("{agent_id}/device/heartbeat-count")
public Uni getHeartbeatCount(
@PathParam("agent_id") Long agentId,
@QueryParam("device_ids") @Separator(",") List deviceIds
) {
return deviceEventService.getHeartbeatCount(agentId, deviceIds)
.invoke(counters ->
logger.debugf(
"Retrieved counters heartbeat: %s",
counters
)
);
}
}
< /code>
@ApplicationScoped
public class DeviceEventService {

private static final Logger logger = Logger.getLogger(
DeviceEventService.class
);

@Inject
DeviceEventRepository deviceEventRepository;

@Inject
DeviceStatusCountRepository deviceStatusCountRepository;

@WithSpan
public Uni getStatusChangeCount(
Long agentId, List deviceIds
) {
return deviceStatusCountRepository.getStatusChangeCount(agentId, deviceIds);

}

@WithSpan
public Uni getHeartbeatCount(
Long agentId, List deviceIds
) {
return deviceStatusCountRepository.getHeartbeatCount(agentId,
deviceIds);

}
}
< /code>
@ApplicationScoped
public class DeviceEventService {

private static final Logger logger = Logger.getLogger(
DeviceEventService.class
);

@Inject
DeviceEventRepository deviceEventRepository;

@Inject
DeviceStatusCountRepository deviceStatusCountRepository;

@WithSpan
public Uni getStatusChangeCount(
Long agentId, List deviceIds
) {
return deviceStatusCountRepository.getStatusChangeCount(agentId, deviceIds);

}

@WithSpan
public Uni getHeartbeatCount(
Long agentId, List deviceIds
) {
return deviceStatusCountRepository.getHeartbeatCount(agentId,
deviceIds);

}
}
< /code>
@ApplicationScoped
public class DeviceStatusCountRepository {

private static final Logger logger = Logger.getLogger(
DeviceStatusCountRepository.class
);

@Inject
@ReactiveDataSource("timescale")
private Pool client;

@WithSpan
public Uni getStatusChangeCount(long agentId, List deviceIds) {
StringBuilder queryBuilder = new StringBuilder("SELECT device_id, up_down_count FROM table_1_with_count WHERE agent_id = $1");
List params = new ArrayList();
params.add(agentId);

if (deviceIds != null && !deviceIds.isEmpty()) {
queryBuilder.append(" AND device_id IN (");
for (int i = 0, j = 2; i < deviceIds.size(); i++, j++) {
if (i >  0) queryBuilder.append(",");
queryBuilder.append("$");
queryBuilder.append(j);
}
queryBuilder.append(")");
params.addAll(deviceIds);
}

String query = queryBuilder.toString();
logger.debugf("Query: %s", query);

return client.withConnection( connection -> connection
.preparedQuery(query)
.execute(Tuple.from(params))
.map(rows -> {
logger.debugf("Query field up_down_count");
List events = new ArrayList();
for (Row row : rows) {
DeviceCountOutDto event = new DeviceCountOutDto(
row.getLong("device_id"),
row.getInteger("up_down_count")
);
events.add(event);
}
return events;
})
);
}

@WithSpan
public Uni getHeartbeatCount(long agentId,
List deviceIds) {
StringBuilder queryBuilder = new StringBuilder("SELECT device_id, heartbeat_count FROM table_2_with_count WHERE agent_id = $1");
List params = new ArrayList();
params.add(agentId);

if (deviceIds != null && !deviceIds.isEmpty()) {
queryBuilder.append(" AND device_id IN (");
for (int i = 0, j = 2; i < deviceIds.size(); i++, j++) {
if (i > 0) queryBuilder.append(",");
queryBuilder.append("$");
queryBuilder.append(j);
}
queryBuilder.append(")");
params.addAll(deviceIds);
}

String query = queryBuilder.toString();
logger.debugf("Query: %s", query);

return client.withConnection( connection -> connection
.preparedQuery(query)
.execute(Tuple.from(params))
.map(rows -> {
logger.debugf("Query field heartbeat_count");
List events = new ArrayList();
for (Row row : rows) {
DeviceCountOutDto event = new DeviceCountOutDto(
row.getLong("device_id"),
row.getInteger("heartbeat_count")
);
events.add(event);
}
return events;
})
);
}
}

Вот список зависимостей:


Подробнее здесь: https://stackoverflow.com/questions/796 ... ed-results