FilesExpand file tree

 master

/

Cli.php

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

99 lines (87 loc) · 3.03 KB

 master

/

Cli.php

Copy path

Top

File metadata and controls

• Code
• Blame

99 lines (87 loc) · 3.03 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

58

59

60

61

62

63

64

65

66

67

68

69

70

71

72

73

74

75

76

77

78

79

80

81

82

83

84

85

86

87

88

89

90

91

92

93

94

95

96

97

98

99

<?php

namespace PHPQueue;

use PHPQueue\Exception\Exception;

class Cli

{

public $queue_name;

public function __construct($options=array())

{

if ( !empty($options['queue']) ) {

$this->queue_name = $options['queue'];

}

}

public function add($payload=array())

{

fwrite(STDOUT, "===========================================================\n");

fwrite(STDOUT, "Adding Job...");

$status = false;

try {

$queue = Base::getQueue($this->queue_name);

$status = Base::addJob($queue, $payload);

fwrite(STDOUT, "Done.\n");

} catch (\Exception $ex) {

fwrite(STDOUT, sprintf("Error: %s\n", $ex->getMessage()));

throw $ex;

}

return $status;

}

public function peek()

{

$newJob = null;

$queue = Base::getQueue($this->queue_name);

try {

$newJob = Base::getJob($queue);

fwrite(STDOUT, "===========================================================\n");

fwrite(STDOUT, "Next Job:\n");

var_dump($newJob);

fwrite(STDOUT, "\nReleasing Job...\n");

$queue->releaseJob($newJob->job_id);

} catch (\Exception $ex) {

fwrite(STDOUT, "Error: " . $ex->getMessage() . "\n");

}

}

public function work()

{

$newJob = null;

$queue = Base::getQueue($this->queue_name);

try {

$newJob = Base::getJob($queue);

fwrite(STDOUT, "===========================================================\n");

fwrite(STDOUT, "Next Job:\n");

var_dump($newJob);

} catch (\Exception $ex) {

fwrite(STDOUT, "Error: " . $ex->getMessage() . "\n");

}

if (empty($newJob)) {

fwrite(STDOUT, "Notice: No Job found.\n");

return;

}

try {

if (empty($newJob->worker)) {

throw new Exception("No worker declared.");

}

if (is_string($newJob->worker)) {

$result_data = $this->processWorker($newJob->worker, $newJob);

} elseif (is_array($newJob->worker)) {

foreach ($newJob->worker as $worker_name) {

$result_data = $this->processWorker($worker_name, $newJob);

$newJob->data = $result_data;

}

}

fwrite(STDOUT, "Updating job... \n");

return Base::updateJob($queue, $newJob->job_id, $result_data);

} catch (\Exception $ex) {

fwrite(STDOUT, sprintf("\nError occured: %s\n", $ex->getMessage()));

$queue->releaseJob($newJob->job_id);

throw $ex;

}

}

protected function processWorker($worker_name, $new_job)

{

fwrite(STDOUT, sprintf("Running worker (%s) now... ", $worker_name));

$newWorker = Base::getWorker($worker_name);

Base::workJob($newWorker, $new_job);

fwrite(STDOUT, "Done.\n");

return $newWorker->result_data;

}

}