Skip to content

Commit b6d60a8

Browse files
author
rezra3
committed
will not fail to read all peers if one has json issue
1 parent 1bf77d8 commit b6d60a8

File tree

1 file changed

+136
-125
lines changed

1 file changed

+136
-125
lines changed

‎src/common/codeine/db/mysql/connectors/StatusMysqlConnector.java‎

Lines changed: 136 additions & 125 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
package codeine.db.mysql.connectors;
22

3+
import com.google.gson.JsonSyntaxException;
34
import java.sql.ResultSet;
45
import java.sql.SQLException;
56
import java.util.List;
@@ -23,133 +24,143 @@
2324
import com.google.common.collect.Maps;
2425
import com.google.gson.Gson;
2526

26-
public class StatusMysqlConnector implements IStatusDatabaseConnector{
27-
private static final Logger log = Logger.getLogger(StatusMysqlConnector.class);
28-
@Inject
29-
private DbUtils dbUtils;
30-
@Inject
31-
private Gson gson;
32-
@Inject private ExperimentalConfJsonStore webConfJsonStore;
33-
private static final String TABLE_NAME = "ProjectStatusList";
34-
35-
36-
37-
public StatusMysqlConnector() {
38-
super();
39-
}
40-
41-
42-
public StatusMysqlConnector(DbUtils dbUtils, Gson gson, ExperimentalConfJsonStore webConfJsonStore) {
43-
super();
44-
this.dbUtils = dbUtils;
45-
this.gson = gson;
46-
this.webConfJsonStore = webConfJsonStore;
47-
}
48-
49-
50-
public void createTables() {
51-
if (webConfJsonStore.get().readonly_web_server()) {
52-
log.info("read only mode");
53-
return;
54-
}
55-
String colsDefinition = "peer_key VARCHAR(150) NOT NULL PRIMARY KEY, data text, update_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP NOT NULL, status VARCHAR(50) DEFAULT 'On' NOT NULL";
56-
dbUtils.executeUpdate("create table if not exists " + TABLE_NAME + " (" + colsDefinition + ")");
57-
}
58-
59-
@Override
60-
public void putReplaceStatus(PeerStatusJsonV2 p) {
61-
String json = gson.toJson(p);
62-
log.info("will update status to " + dbUtils.server() + "\n" + json);
63-
dbUtils.executeUpdate("DELETE FROM "+TABLE_NAME+" WHERE peer_key = '" + p.peer_key() + "'");
64-
dbUtils.executeUpdate("REPLACE INTO "+TABLE_NAME+" (peer_key, data, update_time ) VALUES (?, ?, CURRENT_TIMESTAMP())", p.peer_key(), json);
65-
}
66-
67-
@Override
68-
public Map<String, PeerStatusJsonV2> getPeersStatus() {
69-
log.info("getPeersStatus " + dbUtils.server());
70-
final Map<String, PeerStatusJsonV2> $ = Maps.newHashMap();
71-
Function<ResultSet, Void> function = rs -> {
72-
try {
73-
String key = rs.getString(1);
74-
String value = rs.getString(2);
75-
PeerStatusJsonV2 peerStatus = gson.fromJson(value, PeerStatusJsonV2.class);
76-
peerStatus.status(PeerStatusString.valueOf(rs.getString("status")));
77-
updateNodesWithPeer(peerStatus);
78-
$.put(key, peerStatus);
79-
return null;
80-
} catch (SQLException e) {
81-
throw ExceptionUtils.asUnchecked(e);
82-
}
83-
};
84-
dbUtils.executeQueryCompressed("select * from " + TABLE_NAME, function);
85-
return $;
86-
}
87-
88-
private void updateNodesWithPeer(PeerStatusJsonV2 peerStatus) {
89-
peerStatus.updateNodesWithPeer();
90-
}
91-
92-
@Override
93-
public void updatePeersStatus(final long timeToRemove, final long timeToDisc) {
94-
final List<String> idToRemove = Lists.newArrayList();
95-
final List<String> idToDisc = Lists.newArrayList();
96-
Function<ResultSet, Void> function = rs -> {
97-
try {
98-
String key = rs.getString("peer_key");
27+
public class StatusMysqlConnector implements IStatusDatabaseConnector {
28+
29+
private static final Logger log = Logger.getLogger(StatusMysqlConnector.class);
30+
@Inject
31+
private DbUtils dbUtils;
32+
@Inject
33+
private Gson gson;
34+
@Inject
35+
private ExperimentalConfJsonStore webConfJsonStore;
36+
private static final String TABLE_NAME = "ProjectStatusList";
37+
38+
39+
public StatusMysqlConnector() {
40+
super();
41+
}
42+
43+
44+
public StatusMysqlConnector(DbUtils dbUtils, Gson gson, ExperimentalConfJsonStore webConfJsonStore) {
45+
super();
46+
this.dbUtils = dbUtils;
47+
this.gson = gson;
48+
this.webConfJsonStore = webConfJsonStore;
49+
}
50+
51+
52+
public void createTables() {
53+
if (webConfJsonStore.get().readonly_web_server()) {
54+
log.info("read only mode");
55+
return;
56+
}
57+
String colsDefinition = "peer_key VARCHAR(150) NOT NULL PRIMARY KEY, data text, update_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP NOT NULL, status VARCHAR(50) DEFAULT 'On' NOT NULL";
58+
dbUtils.executeUpdate("create table if not exists " + TABLE_NAME + " (" + colsDefinition + ")");
59+
}
60+
61+
@Override
62+
public void putReplaceStatus(PeerStatusJsonV2 p) {
63+
String json = gson.toJson(p);
64+
log.info("will update status to " + dbUtils.server() + "\n" + json);
65+
dbUtils.executeUpdate("DELETE FROM " + TABLE_NAME + " WHERE peer_key = '" + p.peer_key() + "'");
66+
dbUtils.executeUpdate(
67+
"REPLACE INTO " + TABLE_NAME + " (peer_key, data, update_time ) VALUES (?, ?, CURRENT_TIMESTAMP())",
68+
p.peer_key(), json);
69+
}
70+
71+
@Override
72+
public Map<String, PeerStatusJsonV2> getPeersStatus() {
73+
log.info("getPeersStatus " + dbUtils.server());
74+
final Map<String, PeerStatusJsonV2> $ = Maps.newHashMap();
75+
Function<ResultSet, Void> function = rs -> {
76+
try {
77+
String key = rs.getString(1);
78+
log.debug("Checking key " + key);
79+
String value = rs.getString(2);
80+
PeerStatusJsonV2 peerStatus = gson.fromJson(value, PeerStatusJsonV2.class);
81+
peerStatus.status(PeerStatusString.valueOf(rs.getString("status")));
82+
updateNodesWithPeer(peerStatus);
83+
$.put(key, peerStatus);
84+
return null;
85+
} catch (SQLException e) {
86+
throw ExceptionUtils.asUnchecked(e);
87+
} catch (JsonSyntaxException e) {
88+
log.error("Got json exception while trying to parse line, will skip this node", e);
89+
return null;
90+
}
91+
};
92+
dbUtils.executeQueryCompressed("select * from " + TABLE_NAME, function);
93+
return $;
94+
}
95+
96+
private void updateNodesWithPeer(PeerStatusJsonV2 peerStatus) {
97+
peerStatus.updateNodesWithPeer();
98+
}
99+
100+
@Override
101+
public void updatePeersStatus(final long timeToRemove, final long timeToDisc) {
102+
final List<String> idToRemove = Lists.newArrayList();
103+
final List<String> idToDisc = Lists.newArrayList();
104+
Function<ResultSet, Void> function = rs -> {
105+
try {
106+
String key = rs.getString("peer_key");
99107
// PeerStatusString status = PeerStatusString.valueOf(rs.getString("status"));
100-
String value = rs.getString("data");
101-
String status = rs.getString("status");
102-
PeerStatusJsonV2 peerStatus = gson.fromJson(value, PeerStatusJsonV2.class);
103-
PeerType peerType = peerStatus.peer_type();
104-
long timeToRemovePeer = peerType == PeerType.Reporter ? timeToRemove + TimeUnit.DAYS.toMinutes(7) : timeToRemove;
105-
long timeToDiscPeer = peerType == PeerType.Reporter ? timeToDisc + TimeUnit.DAYS.toMinutes(7) : timeToDisc;
106-
long timeDiff = rs.getLong("TIME_DIFF");
107-
log.debug("time diff is " + timeDiff);
108-
if (timeDiff > timeToRemovePeer){
109-
log.info("time diff is " + timeDiff);
110-
log.info("deleting " + peerStatus);
108+
String value = rs.getString("data");
109+
String status = rs.getString("status");
110+
PeerStatusJsonV2 peerStatus = gson.fromJson(value, PeerStatusJsonV2.class);
111+
PeerType peerType = peerStatus.peer_type();
112+
long timeToRemovePeer =
113+
peerType == PeerType.Reporter ? timeToRemove + TimeUnit.DAYS.toMinutes(7) : timeToRemove;
114+
long timeToDiscPeer =
115+
peerType == PeerType.Reporter ? timeToDisc + TimeUnit.DAYS.toMinutes(7) : timeToDisc;
116+
long timeDiff = rs.getLong("TIME_DIFF");
117+
log.debug("time diff is " + timeDiff);
118+
if (timeDiff > timeToRemovePeer) {
119+
log.info("time diff is " + timeDiff);
120+
log.info("deleting " + peerStatus);
111121
// rs.deleteRow();
112-
idToRemove.add(key);
113-
}
114-
else if (timeDiff > timeToDiscPeer && !status.equals(PeerStatusString.Disc.toString())){
115-
log.info("time diff is " + timeDiff);
116-
log.info("update to disc " + peerStatus);
117-
idToDisc.add(key);
122+
idToRemove.add(key);
123+
} else if (timeDiff > timeToDiscPeer && !status.equals(PeerStatusString.Disc.toString())) {
124+
log.info("time diff is " + timeDiff);
125+
log.info("update to disc " + peerStatus);
126+
idToDisc.add(key);
118127
// rs.updateString("status", "Disc");
119128
// rs.updateRow();
120-
}
121-
return null;
122-
} catch (SQLException e) {
123-
throw ExceptionUtils.asUnchecked(e);
124-
}
125-
};
126-
dbUtils.executeUpdateableQuery("select *,TIMESTAMPDIFF(MINUTE,update_time,CURRENT_TIMESTAMP()) as TIME_DIFF from " + TABLE_NAME, function);
127-
if (webConfJsonStore.get().readonly_web_server()) {
128-
log.info("read only mode");
129-
return;
130-
}
131-
for (String key : idToRemove) {
132-
log.info("deleting " + key);
133-
dbUtils.executeUpdate("DELETE from " + TABLE_NAME + " WHERE peer_key = ?", key);
134-
}
135-
for (String key : idToDisc) {
136-
log.info("discing " + key);
137-
dbUtils.executeUpdate("UPDATE " + TABLE_NAME + " SET status = '" + PeerStatusString.Disc.toString() + "' WHERE peer_key = ?", key);
138-
}
139-
}
140-
141-
142-
@Override
143-
public String server() {
144-
return dbUtils.toString();
145-
}
146-
147-
148-
@Override
149-
public String toString() {
150-
return "StatusMysqlConnector [dbUtils=" + dbUtils + "]";
151-
}
152-
153-
154-
129+
}
130+
return null;
131+
} catch (SQLException e) {
132+
throw ExceptionUtils.asUnchecked(e);
133+
}
134+
};
135+
dbUtils.executeUpdateableQuery(
136+
"select *,TIMESTAMPDIFF(MINUTE,update_time,CURRENT_TIMESTAMP()) as TIME_DIFF from " + TABLE_NAME, function);
137+
if (webConfJsonStore.get().readonly_web_server()) {
138+
log.info("read only mode");
139+
return;
140+
}
141+
for (String key : idToRemove) {
142+
log.info("deleting " + key);
143+
dbUtils.executeUpdate("DELETE from " + TABLE_NAME + " WHERE peer_key = ?", key);
144+
}
145+
for (String key : idToDisc) {
146+
log.info("discing " + key);
147+
dbUtils.executeUpdate(
148+
"UPDATE " + TABLE_NAME + " SET status = '" + PeerStatusString.Disc.toString() + "' WHERE peer_key = ?",
149+
key);
150+
}
151+
}
152+
153+
154+
@Override
155+
public String server() {
156+
return dbUtils.toString();
157+
}
158+
159+
160+
@Override
161+
public String toString() {
162+
return "StatusMysqlConnector [dbUtils=" + dbUtils + "]";
163+
}
164+
165+
155166
}

0 commit comments

Comments
 (0)