-
Notifications
You must be signed in to change notification settings - Fork 1
/
Subscriber.java
81 lines (68 loc) · 2.57 KB
/
Subscriber.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
package pubsub;
import com.impossibl.postgres.api.jdbc.PGConnection;
import com.impossibl.postgres.api.jdbc.PGNotificationListener;
import com.impossibl.postgres.jdbc.PGDataSource;
import data.DbSessionInterface;
import lombok.Getter;
import lombok.Setter;
import util.Labels;
import javax.inject.Inject;
import javax.inject.Singleton;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
@Singleton
public class Subscriber {
private DbSessionInterface dbSession;
private String channelName = Labels.getLabel("pubsub.channel");
private List<AppCallback> list = Collections.synchronizedList(new ArrayList());
@Getter @Setter
private boolean needUpdate = false;
@Inject
public Subscriber(DbSessionInterface dbSession) {
this.dbSession = dbSession;
}
public void initialize(){
Thread dt = new Thread(new RunningThread(), "RunningThread");
dt.setDaemon(true);
dt.start();
}
class RunningThread implements Runnable{
@Override
public void run() {
while(true){
processing();
}
}
private void processing() {
PGNotificationListener listener = new PGNotificationListener() {
@Override
public void notification(int processId, String channelName, String payload) {
// fire action
for (AppCallback e : list) e.fireAction();
}
};
//connection
PGDataSource dataSource = new PGDataSource();
dataSource.setHost("192.168.166.166");
dataSource.setPort(5432);
dataSource.setDatabaseName("postgres");
dataSource.setUser("postgres");
dataSource.setPassword("postgres");
try (PGConnection connection = (PGConnection) dataSource.getConnection()){
Statement statement = connection.createStatement();
statement.execute("LISTEN " + channelName);
statement.close();
connection.addNotificationListener(listener);
Thread.sleep(Long.MAX_VALUE);
} catch (Exception e) {
e.printStackTrace();
}
}
}
public boolean removeObjectForCallback(AppCallback l) {
return list.remove(l);
}
public void addObjectForCallback(AppCallback l) { list.add(l); }
}