Refactored report format and removed redis

com/busdrone/BusReport.java

@@ -1,5 +1,7 @@
 package com.busdrone;
 
+import redis.clients.jedis.Jedis;
+
 public class BusReport {
  String uid;
  String dataProvider;
@@ -24,32 +26,67 @@ public class BusReport {
  long timestamp = java.lang.Long.MIN_VALUE;
  long age = java.lang.Long.MIN_VALUE;
 
- public BusReport cleanup() {
+ transient Jedis db;
+
+ public void cleanup() {
  if (vehicleId == null) vehicleId = coach;
  if (coach == null) coach = vehicleId;
  if (uid == null) uid = dataProvider+'/'+vehicleId;
  if (speedMph == 0 && speedKmh != 0) speedMph = (int) (0.621371 * speedKmh);
  if (speedKmh == 0 && speedMph != 0) speedKmh = (int) (1.609344 * speedMph);
  //if (timestamp = java.lang.Long.MIN_VALUE) 
+ }
+
+ public Event toEvent() {
+ cleanup();
+ Event event = new Event("update_vehicle");
+ event.vehicle = this;
+ return event;
+ }
+
+ public String toEventJson() {
+ return toEvent().toJson();
+ }
+
+ public String syncAndDump(EventStore eventStore) { 
+ cleanup();
+ String key = "com.busdrone.reports/"+uid;
 
+ BusReport oldBus = (BusReport)eventStore.get(key);
 
- return this;
+ if (oldBus == null || !oldBus.equals(this)) {
+ eventStore.put(key, this);
+ return this.toEventJson();
+ } else {
+ return null;
+ }
  }
-
- @Override public boolean equals(Object aThat) {
+ 
+ @Override public boolean equals(Object aThat) { 
  if ( this == aThat ) return true;
  if ( !(aThat instanceof BusReport) ) return false;
  BusReport that = (BusReport)aThat;
 
- return (
+ //System.out.println("equals? "+this.uid+":["+this.lat+","+this.lon+"] "+that.uid+": ["+that.lat+","+that.lon+"]");
+
+ boolean retValx = (
  this.vehicleId == that.vehicleId &&
  this.coach == that.coach &&
  this.routeId == that.routeId &&
  this.route == that.route &&
  this.tripId == that.tripId &&
  this.destination == that.destination &&
- Double.doubleToLongBits(this.lat) == Double.doubleToLongBits(that.lat) &&
- Double.doubleToLongBits(this.lon) == Double.doubleToLongBits(that.lon) &&
- Double.doubleToLongBits(this.heading) == Double.doubleToLongBits(that.heading));
+ (Math.abs(this.lat-that.lat) < 0.00001) &&
+ (Math.abs(this.lon-that.lon) < 0.00001) &&
+ (Math.abs(this.heading-that.heading) < 1));
+
+ boolean retVal = (
+ (Math.abs(this.lat-that.lat) < 0.00001) &&
+ (Math.abs(this.lon-that.lon) < 0.00001) &&
+ (Math.abs(this.heading-that.heading) < 1));
+
+
+ //System.out.println("=>"+retVal);
+ return retVal;
  }
 }

com/busdrone/BusReportServer.java

@@ -4,6 +4,8 @@
 import java.net.InetSocketAddress;
 import java.net.UnknownHostException;
 import java.util.Collection;
+import java.util.HashMap;
+import java.util.Map;
 
 import org.java_websocket.WebSocket;
 import org.java_websocket.WebSocketImpl;
@@ -18,11 +20,11 @@
 import com.busdrone.BusViewFetcher;
 
 public class BusReportServer extends WebSocketServer {
- public JedisPool jedisPool = new JedisPool(new JedisPoolConfig(), "localhost");
+ public EventStore eventStore = new EventStore();
 
  public static void main( String[] args ) throws InterruptedException , IOException {
  WebSocketImpl.DEBUG = false;
- int port = 28737;
+ int port = 28738;
  try {
  port = Integer.parseInt( args[ 0 ] );
  } catch ( Exception ex ) {
@@ -38,7 +40,8 @@ public static void main( String[] args ) throws InterruptedException , IOExcepti
  }
 
  public void sendToAll( String text ) {
- //System.out.println(text);
+ if (text == null) return;
+ System.out.println(text);
  Collection<WebSocket> con = connections();
  synchronized ( con ) {
  for( WebSocket c : con ) {
@@ -64,19 +67,7 @@ public void onMessage( WebSocket conn, String message ) {
  @Override
  public void onOpen(WebSocket conn, ClientHandshake handshake) { 
  System.out.println( "new connection: " + handshake.getResourceDescriptor() );
- /*for (String key : busReportDb.hkeys("buses")) {
- //String foo = busReportDb.get(key);
- //System.out.println(key + ": " + foo);
- //conn.send(busReportDb.get(key));
- System.out.println("Dumping stored set");
- synchronized (conn) {
- conn.send(busReportDb.get(key));
- }
- }*/
-
- Jedis db = jedisPool.getResource();
- try { 
- StringBuilder builder = new StringBuilder();
+ /* StringBuilder builder = new StringBuilder();
  builder.append("["); 
  for (String key : db.hkeys("buses")) {
  if (builder.length() > 1) builder.append(",");
@@ -88,10 +79,7 @@ public void onOpen(WebSocket conn, ClientHandshake handshake) {
  conn.send(db.get("com.busdrone.reports.nextbus"));
  conn.send(db.get("com.busdrone.reports.wsferry"));
  conn.send(db.get("com.busdrone.reports.onebusaway"));
- }
- } finally {
- jedisPool.returnResource(db);
- }
+ */
  }
 
  @Override

com/busdrone/Fetcher.java

@@ -1,24 +1,17 @@
 package com.busdrone;
 
-import redis.clients.jedis.Jedis;
-import redis.clients.jedis.JedisPool;
-import redis.clients.jedis.JedisPoolConfig;
-
 public abstract class Fetcher extends Thread {
  int sleepSecs = 0;
  BusReportServer server;
- JedisPool jedisPool = new JedisPool(new JedisPoolConfig(), "localhost");
-
+
  @Override
  public void run() {
  while (true) {
- Jedis db = jedisPool.getResource();
  try {
- this.runOnce(db);
+ this.runOnce(server.eventStore);
  } catch (Exception e) {
  e.printStackTrace();
  } finally {
- jedisPool.returnResource(db);
  try {
  Thread.sleep(this.sleepSecs * 1000);
  } catch (InterruptedException e) {
@@ -28,5 +21,5 @@ public void run() {
  }
  }
 
- public abstract void runOnce(Jedis db) throws Exception;
+ public abstract void runOnce(EventStore eventStore) throws Exception;
 }

com/busdrone/NextBusFetcher.java

@@ -28,8 +28,8 @@ public NextBusFetcher(BusReportServer s) {
  }
 
  @Override
- public void runOnce(Jedis db) throws Exception {
- ArrayList<BusReport> reports = new ArrayList<BusReport>();
+ public void runOnce(EventStore eventStore) throws Exception {
+ //ArrayList<BusReport> reports = new ArrayList<BusReport>();
 
  Builder parser = new Builder();
  Document doc = parser.build(endpointUrl);
@@ -64,14 +64,16 @@ public void runOnce(Jedis db) throws Exception {
  report.color = report.color + "";
  report.route = report.route + "";
 
- reports.add(report.cleanup());
+ //reports.add(report.cleanup());
  //busReports.put(report.vehicleId, report);
+ //server.sendToAll(report.toEventJson());
+ server.sendToAll(report.syncAndDump(eventStore));
  } catch (Exception e) {
-
+ e.printStackTrace();
  }
  }
 
- String json = gson.toJson(reports.toArray());
- server.sendToAll(json);
+ //String json = gson.toJson(reports.toArray());
+ //server.sendToAll(json);
  }
 }

com/busdrone/OBAFetcher.java

@@ -37,7 +37,7 @@ public OBAFetcher(BusReportServer s) {
  }
 
  @Override
- public void runOnce(Jedis db) throws Exception {
+ public void runOnce(EventStore eventStore) throws Exception {
  int updated=0;
  ArrayList<BusReport> reports = new ArrayList<BusReport>();
  Builder parser = new Builder();
@@ -86,12 +86,12 @@ public void runOnce(Jedis db) throws Exception {
  report.timestamp = Long.parseLong(vehicleStatus.query("lastUpdateTime").get(0).getValue());
  report.age = reportTimestamp - report.timestamp;
 
- if (runCount == 0 || !report.equals(busReports.get(report.vehicleId))) {
- reports.add(report.cleanup()); updated++;
- }
+ //if (runCount == 0 || !report.equals(busReports.get(report.vehicleId))) {
+ // reports.add(report.cleanup()); updated++;
+ //}
 
- busReports.put(report.vehicleId, report);
-
+ //busReports.put(report.vehicleId, report);
+ server.sendToAll(report.syncAndDump(eventStore));
 
  } catch (Exception e) {
  //System.out.println(vehicleStatuses.get(i));
@@ -101,13 +101,13 @@ public void runOnce(Jedis db) throws Exception {
  }
  }
 
- String json = gson.toJson(reports.toArray());
- server.sendToAll(json);
+ //String json = gson.toJson(reports.toArray());
+ //server.sendToAll(json);
 
- if (runCount == 0)
- db.set("com.busdrone.reports.onebusaway", JsonWriter.objectToJson(reports.toArray()));
+ //if (runCount == 0)
+ // db.set("com.busdrone.reports.onebusaway", JsonWriter.objectToJson(reports.toArray()));
 
- if (++runCount >= refreshReferenceInterval) runCount = 0;
+ //if (++runCount >= refreshReferenceInterval) runCount = 0;
 
  //System.out.println("OBAFetcher complete. "+updated+" updated");
  }

com/busdrone/WSFerryFetcher.java

@@ -38,7 +38,7 @@ public WSFerryFetcher(BusReportServer s) {
  }
 
  @Override
- public void runOnce(Jedis db) throws Exception {
+ public void runOnce(EventStore eventStore) throws Exception {
  URLConnection connection = new URL(endpointUrl).openConnection();
  ArrayList<BusReport> reports = new ArrayList<BusReport>();
  JsonParser parser = new JsonParser();
@@ -71,15 +71,18 @@ public void runOnce(Jedis db) throws Exception {
  report.heading = o.get("head").getAsInt();
  report.timestamp = parseDatetime(o.get("datetime").getAsString());
 
- reports.add(report.cleanup());
+ //reports.add(report.cleanup());
+ //server.sendToAll(report.toEventJson());
+
+ server.sendToAll(report.syncAndDump(eventStore));
 
  } catch (Exception e) {}
 
  }
 
- String json = gson.toJson(reports.toArray());
- server.sendToAll(json);
- db.set("com.busdrone.reports.wsferry", json);
+ //String json = gson.toJson(reports.toArray());
+ //server.sendToAll(json);
+ //db.set("com.busdrone.reports.wsferry", json);
  }
 
  @SuppressWarnings("deprecation")