Commit 35516cce authored by Cristiano Urban's avatar Cristiano Urban
Browse files

Added some modifications to test a simple integration of the preprocessor class. 



Signed-off-by: default avatarCristiano Urban <cristiano.urban@inaf.it>
parent 039e4d20

transfer_service/store_amqp_server.py

+3 −0
Original line number Diff line number Diff line
@@ -13,6 +13,7 @@ import json 
from amqp_server import AMQPServer

from job import Job

from job_cache import JobCache

from job_queue import JobQueue

from config import Config
@@ -26,6 +27,7 @@ class StoreAMQPServer(AMQPServer): 
        self.jobCache = JobCache(self.params["host"], 

                                 self.params.getint("port"), 

                                 self.params.getint("db_write"))

        self.pendingQueueWrite = JobQueue("pending")        

        self.job = None

        self.username = None

        self.path = None
@@ -74,6 +76,7 @@ class StoreAMQPServer(AMQPServer): 
                self.prepare(user)

                self.jobCache.set(self.job)

                redisResponse = self.jobCache.get(self.job.jobId)

                self.pendingQueueWrite.insertJob(self.job)

                if "error" in redisResponse:

                    response = { "responseType": "ERROR",

                                 "errorCode": 5,

transfer_service/store_preprocessor.py

+3 −3
Original line number Diff line number Diff line
@@ -145,6 +145,6 @@ class StorePreprocessor(object): 
        self.dbConn.disconnect()



# Test

sp = StorePreprocessor()

sp.prepare("curban")

sp.start()
 
 No newline at end of file
 
#sp = StorePreprocessor()

#sp.prepare("curban")

#sp.start()
 
 No newline at end of file

transfer_service/transfer_service.py

+5 −0
Original line number Diff line number Diff line
@@ -2,6 +2,7 @@ import time 
import os



from job_handler import JobHandler

from preprocessor import Preprocessor

from config import Config
@@ -11,6 +12,7 @@ class TransferService(object): 
        config = Config("vos_ts.conf")

        self.params = config.loadSection("amqp")

        self.jobHandler = JobHandler(self.params["host"])

        self.preprocessor = Preprocessor()



        # PullFromVOSpace

        self.jobHandler.addAMQPServer('start', 'start_job_queue')
@@ -20,6 +22,9 @@ class TransferService(object): 
        # Push

        self.jobHandler.addAMQPServer('store', 'store_job_queue')

        

        # Preprocessor

        self.preprocessor.start()



    def run(self):

        self.jobHandler.run()