bejancsaba commented on code in PR #6733:
URL: https://github.com/apache/nifi/pull/6733#discussion_r1067429017
##########
c2/c2-client-bundle/c2-client-service/src/main/java/org/apache/nifi/c2/client/service/C2ClientService.java:
##########
@@ -32,23 +42,65 @@ public class C2ClientService {
private final C2Client client;
private final C2HeartbeatFactory c2HeartbeatFactory;
- private final C2OperationService operationService;
+ private final C2OperationHandlerProvider operationService;
+ private final RequestedOperationDAO requestedOperationDAO;
+ private final Consumer<C2Operation> c2OperationRegister;
+ private volatile boolean heartbeatLocked = false;
- public C2ClientService(C2Client client, C2HeartbeatFactory
c2HeartbeatFactory, C2OperationService operationService) {
+ public C2ClientService(C2Client client, C2HeartbeatFactory
c2HeartbeatFactory, C2OperationHandlerProvider operationService,
+ RequestedOperationDAO requestedOperationDAO, Consumer<C2Operation>
c2OperationRegister) {
this.client = client;
this.c2HeartbeatFactory = c2HeartbeatFactory;
this.operationService = operationService;
+ this.requestedOperationDAO = requestedOperationDAO;
+ this.c2OperationRegister = c2OperationRegister;
}
public void sendHeartbeat(RuntimeInfoWrapper runtimeInfoWrapper) {
try {
- C2Heartbeat c2Heartbeat =
c2HeartbeatFactory.create(runtimeInfoWrapper);
-
client.publishHeartbeat(c2Heartbeat).ifPresent(this::processResponse);
+ if (heartbeatLocked) {
+ logger.debug("Restart is in progress, skipping heartbeat");
+ } else {
+ C2Heartbeat c2Heartbeat =
c2HeartbeatFactory.create(runtimeInfoWrapper);
+
client.publishHeartbeat(c2Heartbeat).ifPresent(this::processResponse);
+ }
} catch (Exception e) {
logger.error("Failed to send/process heartbeat:", e);
}
}
+ public void sendAcknowledge(C2OperationAck operationAck) {
+ try {
+ client.acknowledgeOperation(operationAck);
+ } catch (Exception e) {
+ logger.error("Failed to send acknowledge:", e);
+ }
+ }
+
+ public void enableHeartbeat() {
+ heartbeatLocked = false;
+ }
+
+ public void handleRequestedOperations(List<C2Operation>
requestedOperations) {
+ LinkedList<C2Operation> lRequestedOperations = new
LinkedList<>(requestedOperations);
+ C2Operation requestedOperation;
+ while ((requestedOperation = lRequestedOperations.poll()) != null) {
Review Comment:
ok, works for me I just thought if it is a queue and we treat it as a queue
and we use it as a queue we should use queue :) but you are right list works as
well.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscr...@nifi.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
