enqueue with options

rpc.go

@@ -27,6 +27,7 @@ import (
 	"github.com/gogo/protobuf/proto"
 	"github.com/topfreegames/pitaya/constants"
 	"github.com/topfreegames/pitaya/route"
+	"github.com/topfreegames/pitaya/worker"
 )
 
 // RPC calls a method in a different server
@@ -40,6 +41,7 @@ func RPCTo(ctx context.Context, serverID, routeStr string, reply proto.Message,
 }
 
 // ReliableRPC enqueues RPC to worker so it's executed asynchronously
+// Default enqueue options are used
 func ReliableRPC(
 	routeStr string,
 	metadata map[string]interface{},
@@ -48,6 +50,17 @@ func ReliableRPC(
 	return app.worker.EnqueueRPC(routeStr, metadata, reply, arg)
 }
 
+// ReliableRPCWithOptions enqueues RPC to worker
+// Receive worker options for this specific RPC
+func ReliableRPCWithOptions(
+	routeStr string,
+	metadata map[string]interface{},
+	reply, arg proto.Message,
+	opts *worker.EnqueueOpts,
+) (jid string, err error) {
+	return app.worker.EnqueueRPCWithOptions(routeStr, metadata, reply, arg, opts)
+}
+
 func doSendRPC(ctx context.Context, serverID, routeStr string, reply proto.Message, arg proto.Message) error {
 	if app.rpcServer == nil {
 		return constants.ErrRPCServerNotInitialized

worker/worker.go

@@ -84,7 +84,7 @@ func (w *Worker) EnqueueRPC(
 	metadata map[string]interface{},
 	reply, arg proto.Message,
 ) (jid string, err error) {
-	opts := w.enqueueOptions()
+	opts := w.enqueueOptions(w.opts)
 	return workers.EnqueueWithOptions(rpcQueue, class, &rpcInfo{
 		Route:    routeStr,
 		Metadata: metadata,
@@ -93,6 +93,21 @@ func (w *Worker) EnqueueRPC(
 	}, opts)
 }
 
+// EnqueueRPCWithOptions enqueues rpc job to worker
+func (w *Worker) EnqueueRPCWithOptions(
+	routeStr string,
+	metadata map[string]interface{},
+	reply, arg proto.Message,
+	opts *EnqueueOpts,
+) (jid string, err error) {
+	return workers.EnqueueWithOptions(rpcQueue, class, &rpcInfo{
+		Route:    routeStr,
+		Metadata: metadata,
+		Arg:      arg,
+		Reply:    reply,
+	}, w.enqueueOptions(opts))
+}
+
 // RegisterRPCJob registers a RPC job
 func (w *Worker) RegisterRPCJob(rpcJob RPCJob) error {
 	if w.registered {
@@ -152,15 +167,17 @@ func (w *Worker) parsedRPCJob(rpcJob RPCJob) func(*workers.Msg) {
 	}
 }
 
-func (w *Worker) enqueueOptions() workers.EnqueueOptions {
+func (w *Worker) enqueueOptions(
+	opts *EnqueueOpts,
+) workers.EnqueueOptions {
 	return workers.EnqueueOptions{
-		Retry:    w.opts.RetryEnabled,
-		RetryMax: w.opts.MaxRetries,
+		Retry:    opts.RetryEnabled,
+		RetryMax: opts.MaxRetries,
 		RetryOptions: workers.RetryOptions{
-			Exp:      w.opts.ExponentialFactor,
-			MinDelay: w.opts.MinDelayToRetry,
-			MaxDelay: w.opts.MaxDelayToRetry,
-			MaxRand:  w.opts.MaxRandom,
+			Exp:      opts.ExponentialFactor,
+			MinDelay: opts.MinDelayToRetry,
+			MaxDelay: opts.MaxDelayToRetry,
+			MaxRand:  opts.MaxRandom,
 		},
 	}
 }