Implement retry times for harvest objects

ckanext/harvest/model/__init__.py

@@ -200,7 +200,7 @@ def define_harvester_tables():
         Column('import_finished', types.DateTime),
         Column('state', types.UnicodeText, default=u'WAITING'),
         Column('metadata_modified_date', types.DateTime),
-        Column('retry_times',types.Integer),
+        Column('retry_times',types.Integer, default=0),
         Column('harvest_job_id', types.UnicodeText, ForeignKey('harvest_job.id')),
         Column('harvest_source_id', types.UnicodeText, ForeignKey('harvest_source.id')),
         Column('package_id', types.UnicodeText, ForeignKey('package.id'), nullable=True),
@@ -375,6 +375,7 @@ ALTER TABLE harvest_object_extra
 	ADD CONSTRAINT harvest_object_extra_harvest_object_id_fkey FOREIGN KEY (harvest_object_id) REFERENCES harvest_object(id);
 
 UPDATE harvest_object set state = 'COMPLETE';
+UPDATE harvest_object set retry_times = 0;
 UPDATE harvest_source set frequency = 'MANUAL';
 
 """

ckanext/harvest/queue.py

@@ -158,6 +158,14 @@ def fetch_callback(channel, method, header, body):
         channel.basic_ack(method.delivery_tag)
         return False
 
+    obj.retry_times += 1
+    obj.save()
+
+    if obj.retry_times >= 5:
+        log.error('Too many consecutive retries for object {0}'.format(obj.id))
+        channel.basic_ack(method.delivery_tag)
+        return False
+
     # Send the harvest object to the plugins that implement
     # the Harvester interface, only if the source type
     # matches
@@ -171,7 +179,6 @@ def fetch_callback(channel, method, header, body):
             success = harvester.fetch_stage(obj)
             obj.fetch_finished = datetime.datetime.now()
             obj.save()
-            #TODO: retry times?
             if success:
                 # If no errors where found, call the import method
                 obj.import_started = datetime.datetime.now()