X-Git-Url: https://git.decadent.org.uk/gitweb/?a=blobdiff_plain;f=dak%2Fimport_known_changes.py;h=4e8068f609a8a00b29cfdce45462477c75ea648c;hb=9a4fb568340b14698947bfa59b309ae0c67c693a;hp=0b2386263143be2fe184d09ba3c50848d7c422e2;hpb=b1ed62bc33000e2670aed2b2f1c478b2e6c62d0e;p=dak.git

diff --git a/dak/import_known_changes.py b/dak/import_known_changes.py
index 0b238626..4e8068f6 100755
--- a/dak/import_known_changes.py
+++ b/dak/import_known_changes.py
@@ -32,7 +32,7 @@ import sys
 import os
 import logging
 import threading
-from daklib.dbconn import DBConn,get_knownchange
+from daklib.dbconn import DBConn, get_dbchange, get_policy_queue
 from daklib.config import Config
 import apt_pkg
 from daklib.dak_exceptions import DBUpdateError, InvalidDscError, ChangesUnicodeError
@@ -70,7 +70,6 @@ def check_signature (sig_filename, data_filename=""):
 
     keyrings = [
         "/home/joerg/keyring/keyrings/debian-keyring.gpg",
-        "/home/joerg/keyring/keyrings/debian-keyring.pgp",
         "/home/joerg/keyring/keyrings/debian-maintainers.gpg",
         "/home/joerg/keyring/keyrings/debian-role-keys.gpg",
         "/home/joerg/keyring/keyrings/emeritus-keyring.pgp",
@@ -128,37 +127,48 @@ class OneAtATime(object):
     """
     def __init__(self):
         self.next_in_line = None
-        self.next_lock = threading.Condition()
+        self.read_lock = threading.Condition()
+        self.write_lock = threading.Condition()
         self.die = False
 
     def plsDie(self):
         self.die = True
-        self.next_lock.notify()
+        self.write_lock.acquire()
+        self.write_lock.notifyAll()
+        self.write_lock.release()
+
+        self.read_lock.acquire()
+        self.read_lock.notifyAll()
+        self.read_lock.release()
 
     def enqueue(self, next):
-        self.next_lock.acquire()
+        self.write_lock.acquire()
         while self.next_in_line:
             if self.die:
                 return
-            self.next_lock.wait()
+            self.write_lock.wait()
 
         assert( not self.next_in_line )
         self.next_in_line = next
-        self.next_lock.notifyAll()
-        self.next_lock.release()
+        self.write_lock.release()
+        self.read_lock.acquire()
+        self.read_lock.notify()
+        self.read_lock.release()
 
     def dequeue(self):
-        self.next_lock.acquire()
+        self.read_lock.acquire()
         while not self.next_in_line:
             if self.die:
                 return
-            self.next_lock.wait()
+            self.read_lock.wait()
 
         result = self.next_in_line
 
         self.next_in_line = None
-        self.next_lock.notifyAll()
-        self.next_lock.release()
+        self.read_lock.release()
+        self.write_lock.acquire()
+        self.write_lock.notify()
+        self.write_lock.release()
 
         if isinstance(result, EndOfChanges):
             return None
@@ -190,8 +200,18 @@ class ChangesGenerator(threading.Thread):
     def run(self):
         cnf = Config()
         count = 1
-        for directory in [ "Accepted", "Byhand", "Done", "New", "ProposedUpdates", "OldProposedUpdates" ]:
-            checkdir = cnf["Dir::Queue::%s" % (directory) ]
+
+        dirs = []
+        dirs.append(cnf['Dir::Done'])
+
+        for queue_name in [ "byhand", "new", "proposedupdates", "oldproposedupdates" ]:
+            queue = get_policy_queue(queue_name)
+            if queue:
+                dirs.append(os.path.abspath(queue.path))
+            else:
+                warn("Could not find queue %s in database" % queue_name)
+
+        for checkdir in dirs:
             if os.path.exists(checkdir):
                 print "Looking into %s" % (checkdir)
 
@@ -207,7 +227,7 @@ class ChangesGenerator(threading.Thread):
                                 continue
                             count += 1
 
-                            if not get_knownchange(changesfile, self.session):
+                            if not get_dbchange(changesfile, self.session):
                                 to_import = ChangesToImport(dirpath, changesfile, count)
                                 if self.die:
                                     return
@@ -246,10 +266,10 @@ class ImportThread(threading.Thread):
                 changesfile = os.path.join(to_import.dirpath, to_import.changesfile)
                 changes.changes = parse_changes(changesfile, signing_rules=-1)
                 changes.changes["fingerprint"] = check_signature(changesfile)
-                changes.add_known_changes(to_import.dirpath, self.session)
+                changes.add_known_changes(to_import.dirpath, session=self.session)
                 self.session.commit()
 
-            except InvalidDscError, line:
+            except InvalidDscError as line:
                 warn("syntax error in .dsc file '%s', line %s." % (f, line))
 
             except ChangesUnicodeError:
@@ -278,7 +298,7 @@ class ImportKnownChanges(object):
 
         except KeyboardInterrupt:
             print("Caught C-c; terminating.")
-            utils.warn("Caught C-c; terminating.")
+            warn("Caught C-c; terminating.")
             self.plsDie()
 
     def plsDie(self):
@@ -301,7 +321,7 @@ def main():
                  ('v',"verbose", "%s::%s" % (options_prefix,"Verbose")),
                 ]
 
-    args = apt_pkg.ParseCommandLine(cnf.Cnf, arguments,sys.argv)
+    args = apt_pkg.parse_commandline(cnf.Cnf, arguments,sys.argv)
 
     num_threads = 1