Commit a10c00ad authored by Cristiano Urban's avatar Cristiano Urban
Browse files

Added 'extractJob()' method + modified 'getJob()' method. 



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

transfer_service/job_queue.py

+25 −3
Original line number Diff line number Diff line
@@ -29,8 +29,16 @@ class JobQueue(object): 
    # Gets a copy of the first job without moving it out from the current 

    # queue

    def getJob(self):

        jobObj = self.redisCli.lrange(self.queueName, self.len() - 1, self.len() - 1)[0].decode("utf-8")

        return json.loads(jobObj)

        job = json.loads(self.redisCli.lrange(self.queueName, self.len() - 1, self.len() - 1)[0].decode("utf-8"))

        jobObj = Job()

        jobObj.setId(job["jobId"])

        jobObj.setType(job["jobType"])

        jobObj.setOwnerId(job["ownerId"])

        jobObj.setPhase(job["phase"])

        jobObj.setQuote(job["quote"])

        jobObj.setExecutionDuration(job["executionDuration"])

        jobObj.setInfo(job["jobInfo"])

        return jobObj



    # Pushes a new job into the queue

    def insertJob(self, jobObj):
@@ -48,10 +56,24 @@ class JobQueue(object): 
                 "jobInfo": jobObj.jobInfo }

        self.redisCli.lpush(self.queueName, json.dumps(data))



    # Moves out a job from the end of the current queue

    def extractJob(self):

        job = json.loads(self.redisCli.brpop(self.queueName)[1].decode("utf-8"))

        jobObj = Job()

        jobObj.setId(job["jobId"])

        jobObj.setType(job["jobType"])

        jobObj.setOwnerId(job["ownerId"])

        jobObj.setPhase(job["phase"])

        jobObj.setQuote(job["quote"])

        jobObj.setExecutionDuration(job["executionDuration"])

        jobObj.setInfo(job["jobInfo"])

        return jobObj



    # Moves out a job from the end of the current queue and places it 

    # at the beginning of the next queue (this operation is atomic)

    def moveJobTo(self, nextQueueName):

        self.redisCli.brpoplpush(self.queueName, nextQueueName, timeout = 0)

        self.redisCli.brpoplpush(self.queueName, nextQueueName)







# Test