-
Notifications
You must be signed in to change notification settings - Fork 897
/
StartStopController.java
84 lines (65 loc) · 2.12 KB
/
StartStopController.java
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
package io.confluent.kpay.control;
import io.confluent.kpay.control.model.Status;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.Topology;
import org.apache.kafka.streams.kstream.KStream;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Properties;
/**
* Starts and stops a Controllable instance
*/
public class StartStopController {
private static final Logger log = LoggerFactory.getLogger(StartStopController.class);
private final Topology topology;
private Properties streamsConfig;
private Controllable controllable;
private KafkaStreams streams;
boolean isProcessing = true;
public StartStopController(String controllerTopic, Properties streamsConfig, Controllable controllable){
this.streamsConfig = streamsConfig;
this.controllable = controllable;
/**
* TODO: store history in a hopping window
*/
StreamsBuilder builder = new StreamsBuilder();
KStream<String, Status> stream = builder.stream(controllerTopic);
stream.foreach((key, value) -> {
if (value.getCode() == Status.Code.PAUSE) {
pause();
} else if (value.getCode() == Status.Code.START) {
resume();
}
});
topology = builder.build();
}
public void resume() {
log.info("Resume:" + controllable);
controllable.startProcessing();
isProcessing = true;
}
public void pause() {
log.info("Pause:" + controllable);
controllable.pauseProcessing();
isProcessing = false;
}
public boolean isProcessing() {
return isProcessing;
}
public void start() {
streams = new KafkaStreams(topology, streamsConfig);
streams.start();
log.info(topology.describe().toString());
}
public void stop() {
streams.close();
streams.cleanUp();
}
public Topology getTopology() {
return topology;
}
public String status() {
return Boolean.toString(isProcessing);
}
}