Voomra/trader
Archived
0
Fork 0
Code Activity

fix broken receive procedure

Browse Source
This commit is contained in:
iMoHax
2014-09-04 20:23:34 +04:00
parent 1e8dce9477
commit df73bdf764
1 changed files with 29 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

54
utils/src/main/java/ru/trader/emdn/EMDN.java
View File

@@ -61,29 +61,6 @@ public class EMDN {
context.term(); context.term();
} }
private void parseCSV(String csv) {
LOG.debug("Parse csv: {}", csv);
if (csv.isEmpty()) return;
String[] flds = csv.split(",");
// buyPrice,sellPrice,demand,demandLevel,stationStock,stationStockLevel,categoryName,itemName,stationName,timestamp
ItemData item = new ItemData(flds[7]);
item.setBuy(Double.valueOf(flds[0]), Long.valueOf(flds[4]));
item.setSell(Double.valueOf(flds[1]), Long.valueOf(flds[2]));
LOG.trace("Item: {}", item);
String stName = flds[8].split("\\(")[0].trim();
LOG.trace("Station: {}", stName);
Station station = cache.get(stName);
if (station != null){
LOG.trace("Is old, update");
station.update(item);
} else {
LOG.trace("Is new, create");
station = new Station(stName);
station.update(item);
cache.put(stName, station);
}
}
private byte[] decompress(byte[] input){ private byte[] decompress(byte[] input){
byte[] decompressed = new byte[input.length * 16]; byte[] decompressed = new byte[input.length * 16];
int decompressedLength = 0; int decompressedLength = 0;
@@ -132,14 +109,19 @@ public class EMDN {
subscriber.connect(subServer); subscriber.connect(subServer);
LOG.trace("Subscribe"); LOG.trace("Subscribe");
subscriber.subscribe(new byte[0]); subscriber.subscribe(new byte[0]);
Station station = null;
while (!executor.isShutdown() && !receive.isCancelled()){ while (!executor.isShutdown() && !receive.isCancelled()){
try { try {
byte[] receivedData = subscriber.recv(0); byte[] receivedData = subscriber.recv(0);
LOG.trace("Received data: {}", receivedData); LOG.trace("Received data: {}", receivedData);
if (receivedData == null) return; if (receivedData == null) continue;
//receivedData = decompress(receivedData); //receivedData = decompress(receivedData);
String market_csv = new String(receivedData, "UTF-8"); String market_csv = new String(receivedData, "UTF-8");
parseCSV(market_csv); station = parseCSV(market_csv, station);
if (!subscriber.hasReceiveMore()){
cache.put(station.getName(), station);
station = null;
}
} catch (ZMQException | UnsupportedEncodingException ex) { } catch (ZMQException | UnsupportedEncodingException ex) {
LOG.error("Error on get data from EMDN", ex); LOG.error("Error on get data from EMDN", ex);
} }
@@ -148,5 +130,27 @@ public class EMDN {
LOG.error("Error on connect to EMDN", ex); LOG.error("Error on connect to EMDN", ex);
} }
} }
private Station parseCSV(String csv, Station station) {
LOG.debug("Parse csv: {}", csv);
if (!csv.isEmpty()) {
String[] flds = csv.split(",");
// buyPrice,sellPrice,demand,demandLevel,stationStock,stationStockLevel,categoryName,itemName,stationName,timestamp
ItemData item = new ItemData(flds[7]);
item.setBuy(Double.valueOf(flds[0]), Long.valueOf(flds[4]));
item.setSell(Double.valueOf(flds[1]), Long.valueOf(flds[2]));
LOG.trace("Item: {}", item);
String stName = flds[8].split("\\(")[0].trim();
LOG.trace("Station: {}", stName);
if (station == null) {
LOG.trace("Is new, create");
station = new Station(stName);
}
assert stName.equals(station.getName());
station.update(item);
}
return station;
}
} }
} }