Don't use commit() on each SQL insert, it makes things terrislow.

apps/modes_rx

@@ -88,23 +88,24 @@ def main():
 
   if options.kml is not None:
     dbname = 'adsb.db'
-    sqldb = air_modes.output_sql(my_position, dbname, context) #input into the db
-    #kmlgen = air_modes.output_kml(options.kml, dbname, my_position, lock) #create a KML generating thread to read from the db
+    lock = threading.Lock()
+    sqldb = air_modes.output_sql(my_position, dbname, lock) #input into the db
+    kmlgen = air_modes.output_kml(options.kml, dbname, my_position, lock) #create a KML generating thread to read from the db
+    relay.subscribe("dl_data", sqldb.insert)
 
   printer = None
   if options.no_print is not True:
     relay.subscribe("dl_data", air_modes.output_print(my_position).output)
-    #printer = screen_printer(my_position, context)
 
-#  if options.multiplayer is not None:
-#    [fghost, fgport] = options.multiplayer.split(':')
-#    fgout = air_modes.output_flightgear(my_position, fghost, int(fgport))
-#    tb.subscribe('dl_data', fgout.output)
+  if options.multiplayer is not None:
+    [fghost, fgport] = options.multiplayer.split(':')
+    fgout = air_modes.output_flightgear(my_position, fghost, int(fgport))
+    relay.subscribe("dl_data", fgout.output)
 
 #  if options.sbs1 is True:
 #    sbs1port = air_modes.output_sbs1(my_position, 30003)
-#    tb.subscribe('dl_data', sbs1port.output)
-    #updates.append(sbs1port.add_pending_conns)
+#    relay.subscribe("dl_data", sbs1port.output)
+    #updates.append(sbs1port.add_pending_conns) #TODO FIXME
   tb.run()
   tb.cleanup()
 
@@ -112,8 +113,8 @@ def main():
   relay.finished.wait(0.2)
 
   if options.kml is not None:
-    sqldb.done = True
-    #sqldb.join()
+    kmlgen.close()
+    
 
 if __name__ == '__main__':
   main()

python/kml.py

@@ -31,19 +31,28 @@ class output_kml(threading.Thread):
         self._timeout = timeout
         self._lock = lock
         
-        self.done = False
+        self.shutdown = threading.Event()
+        self.finished = threading.Event()
         self.setDaemon(1)
         self.start()
 
     def run(self):
         self._db = sqlite3.connect(self._dbname) #read from the db
-        while self.done is False:
+        while self.shutdown.is_set() is False:
             self.writekml()
             time.sleep(self._timeout)
                 
-        self.done = True
         self._db.close()
         self._db = None
+        self.finished.set()
+
+    def close(self):
+        self.shutdown.set()
+        self.finished.wait(0.2)
+        #there's a bug here where self._timeout is long and close() has
+        #to wait for the sleep to expire before closing. we just bail
+        #instead with the 0.2 param above.
+        
         
     def writekml(self):
         kmlstr = self.genkml()

python/sql.py

@@ -26,20 +26,11 @@ import sqlite3
 from air_modes.exceptions import *
 import zmq
 
-class output_sql(air_modes.parse, threading.Thread):
-  def __init__(self, mypos, filename, context, addr=None, port=None):
-    threading.Thread.__init__(self)
+class output_sql(air_modes.parse):
+  def __init__(self, mypos, filename, lock, addr=None):
     air_modes.parse.__init__(self, mypos)
 
-    #init socket
-    self._subscriber = context.socket(zmq.SUB)
-    if addr is not None:
-        self._subscriber.connect("tcp://%s:%i" % (addr, port))
-    else:
-        self._subscriber.connect("inproc://modes-radio-pub")
-    self._subscriber.setsockopt(zmq.SUBSCRIBE, "dl_data")
-
-    self._lock = threading.Lock()
+    self._lock = lock;
     #create the database
     self.filename = filename
     self._db = sqlite3.connect(filename)
@@ -72,21 +63,6 @@ class output_sql(air_modes.parse, threading.Thread):
     self._db.close()
     self._db = None
 
-    self.setDaemon(True)
-    self.done = False
-    self.start()
-
-  def run(self):
-    while not self.done:
-        [address, msg] = self._subscriber.recv_multipart() #blocking
-        try:
-            self.insert(msg)
-        except ADSBError:
-            pass
-
-    self._subscriber.close()
-    self._db = None
-
   def insert(self, message):
     with self._lock:
       try:
@@ -102,7 +78,7 @@ class output_sql(air_modes.parse, threading.Thread):
             c = self._db.cursor()
             c.execute(query)
             c.close()
-            self._db.commit()
+#            self._db.commit()
 
       except ADSBError:
         pass