4 files changed, 234 insertions, 0 deletions
diff --git a/plugins/jobs/structs/job.go b/plugins/jobs/structs/job.go
new file mode 100644
index 00000000..2e394543
--- /dev/null
+++ b/plugins/jobs/structs/job.go
@@ -0,0 +1,35 @@
+package structs
+
+import (
+	json "github.com/json-iterator/go"
+	"github.com/spiral/roadrunner/v2/utils"
+)
+
+// Job carries information about single job.
+type Job struct {
+	// Job contains name of job broker (usually PHP class).
+	Job string `json:"job"`
+
+	// Payload is string data (usually JSON) passed to Job broker.
+	Payload string `json:"payload"`
+
+	// Options contains set of PipelineOptions specific to job execution. Can be empty.
+	Options *Options `json:"options,omitempty"`
+}
+
+// Body packs job payload into binary payload.
+func (j *Job) Body() []byte {
+	return utils.AsBytes(j.Payload)
+}
+
+// Context packs job context (job, id) into binary payload.
+func (j *Job) Context(id string) []byte {
+	ctx, _ := json.Marshal(
+		struct {
+			ID  string `json:"id"`
+			Job string `json:"job"`
+		}{ID: id, Job: j.Job},
+	)
+
+	return ctx
+}
diff --git a/plugins/jobs/structs/job_options.go b/plugins/jobs/structs/job_options.go
new file mode 100644
index 00000000..1507d053
--- /dev/null
+++ b/plugins/jobs/structs/job_options.go
@@ -0,0 +1,70 @@
+package structs
+
+import "time"
+
+// Options carry information about how to handle given job.
+type Options struct {
+	// Pipeline manually specified pipeline.
+	Pipeline string `json:"pipeline,omitempty"`
+
+	// Delay defines time duration to delay execution for. Defaults to none.
+	Delay int `json:"delay,omitempty"`
+
+	// Attempts define maximum job retries. Attention, value 1 will only allow job to execute once (without retry).
+	// Minimum valuable value is 2.
+	Attempts int `json:"maxAttempts,omitempty"`
+
+	// RetryDelay defines for how long job should be waiting until next retry. Defaults to none.
+	RetryDelay int `json:"retryDelay,omitempty"`
+
+	// Reserve defines for how broker should wait until treating job are failed. Defaults to 30 min.
+	Timeout int `json:"timeout,omitempty"`
+}
+
+// Merge merges job options.
+func (o *Options) Merge(from *Options) {
+	if o.Pipeline == "" {
+		o.Pipeline = from.Pipeline
+	}
+
+	if o.Attempts == 0 {
+		o.Attempts = from.Attempts
+	}
+
+	if o.Timeout == 0 {
+		o.Timeout = from.Timeout
+	}
+
+	if o.RetryDelay == 0 {
+		o.RetryDelay = from.RetryDelay
+	}
+
+	if o.Delay == 0 {
+		o.Delay = from.Delay
+	}
+}
+
+// CanRetry must return true if broker is allowed to re-run the job.
+func (o *Options) CanRetry(attempt int) bool {
+	// Attempts 1 and 0 has identical effect
+	return o.Attempts > (attempt + 1)
+}
+
+// RetryDuration returns retry delay duration in a form of time.Duration.
+func (o *Options) RetryDuration() time.Duration {
+	return time.Second * time.Duration(o.RetryDelay)
+}
+
+// DelayDuration returns delay duration in a form of time.Duration.
+func (o *Options) DelayDuration() time.Duration {
+	return time.Second * time.Duration(o.Delay)
+}
+
+// TimeoutDuration returns timeout duration in a form of time.Duration.
+func (o *Options) TimeoutDuration() time.Duration {
+	if o.Timeout == 0 {
+		return 30 * time.Minute
+	}
+
+	return time.Second * time.Duration(o.Timeout)
+}
diff --git a/plugins/jobs/structs/job_options_test.go b/plugins/jobs/structs/job_options_test.go
new file mode 100644
index 00000000..18702394
--- /dev/null
+++ b/plugins/jobs/structs/job_options_test.go
@@ -0,0 +1,110 @@
+package structs
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestOptions_CanRetry(t *testing.T) {
+	opts := &Options{Attempts: 0}
+
+	assert.False(t, opts.CanRetry(0))
+	assert.False(t, opts.CanRetry(1))
+}
+
+func TestOptions_CanRetry_SameValue(t *testing.T) {
+	opts := &Options{Attempts: 1}
+
+	assert.False(t, opts.CanRetry(0))
+	assert.False(t, opts.CanRetry(1))
+}
+
+func TestOptions_CanRetry_Value(t *testing.T) {
+	opts := &Options{Attempts: 2}
+
+	assert.True(t, opts.CanRetry(0))
+	assert.False(t, opts.CanRetry(1))
+	assert.False(t, opts.CanRetry(2))
+}
+
+func TestOptions_CanRetry_Value3(t *testing.T) {
+	opts := &Options{Attempts: 3}
+
+	assert.True(t, opts.CanRetry(0))
+	assert.True(t, opts.CanRetry(1))
+	assert.False(t, opts.CanRetry(2))
+}
+
+func TestOptions_RetryDuration(t *testing.T) {
+	opts := &Options{RetryDelay: 0}
+	assert.Equal(t, time.Duration(0), opts.RetryDuration())
+}
+
+func TestOptions_RetryDuration2(t *testing.T) {
+	opts := &Options{RetryDelay: 1}
+	assert.Equal(t, time.Second, opts.RetryDuration())
+}
+
+func TestOptions_DelayDuration(t *testing.T) {
+	opts := &Options{Delay: 0}
+	assert.Equal(t, time.Duration(0), opts.DelayDuration())
+}
+
+func TestOptions_DelayDuration2(t *testing.T) {
+	opts := &Options{Delay: 1}
+	assert.Equal(t, time.Second, opts.DelayDuration())
+}
+
+func TestOptions_TimeoutDuration(t *testing.T) {
+	opts := &Options{Timeout: 0}
+	assert.Equal(t, time.Minute*30, opts.TimeoutDuration())
+}
+
+func TestOptions_TimeoutDuration2(t *testing.T) {
+	opts := &Options{Timeout: 1}
+	assert.Equal(t, time.Second, opts.TimeoutDuration())
+}
+
+func TestOptions_Merge(t *testing.T) {
+	opts := &Options{}
+
+	opts.Merge(&Options{
+		Pipeline:   "pipeline",
+		Delay:      2,
+		Timeout:    1,
+		Attempts:   1,
+		RetryDelay: 1,
+	})
+
+	assert.Equal(t, "pipeline", opts.Pipeline)
+	assert.Equal(t, 1, opts.Attempts)
+	assert.Equal(t, 2, opts.Delay)
+	assert.Equal(t, 1, opts.Timeout)
+	assert.Equal(t, 1, opts.RetryDelay)
+}
+
+func TestOptions_MergeKeepOriginal(t *testing.T) {
+	opts := &Options{
+		Pipeline:   "default",
+		Delay:      10,
+		Timeout:    10,
+		Attempts:   10,
+		RetryDelay: 10,
+	}
+
+	opts.Merge(&Options{
+		Pipeline:   "pipeline",
+		Delay:      2,
+		Timeout:    1,
+		Attempts:   1,
+		RetryDelay: 1,
+	})
+
+	assert.Equal(t, "default", opts.Pipeline)
+	assert.Equal(t, 10, opts.Attempts)
+	assert.Equal(t, 10, opts.Delay)
+	assert.Equal(t, 10, opts.Timeout)
+	assert.Equal(t, 10, opts.RetryDelay)
+}
diff --git a/plugins/jobs/structs/job_test.go b/plugins/jobs/structs/job_test.go
new file mode 100644
index 00000000..e7240c6b
--- /dev/null
+++ b/plugins/jobs/structs/job_test.go
@@ -0,0 +1,19 @@
+package structs
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestJob_Body(t *testing.T) {
+	j := &Job{Payload: "hello"}
+
+	assert.Equal(t, []byte("hello"), j.Body())
+}
+
+func TestJob_Context(t *testing.T) {
+	j := &Job{Job: "job"}
+
+	assert.Equal(t, []byte(`{"id":"id","job":"job"}`), j.Context("id"))
+}