gosubscriber

gosubscriber is a Resque Bus-compatible, Go-based background subscriber build on top of Benjamin Manns' goworker. It allows you to publish Resque Bus events in Ruby (or any other language), and perform background tasks in Go.

Gosubscriber subscribers can run alongside Ruby/Node.js ResqueBus workers and subscribe to the exact same events, so a single action can prompt any number of orchestrated reactions across any number of servers.

Installation

To install gosubscriber, use

go get github.com/delectable/gosubscriber

to install the package, and then from your worker

import "github.com/delectable/gosubscriber"

Getting Started

To create a worker, write a function matching the signature

func(args map[string]interface{}) error

and subscribe it to an event using:

gosubscriber.Subscribe("my_application", "my_queue", "my_event", mySubscriber)

Here is a simple subscriber that subscribes to the event testEventOne and prints its arguments:

package main

import (
  "fmt"
  "github.com/delectable/gosubscriber"
)

func testSubscriber(args map[string]interface{}) error {
  fmt.Printf("testSubscriber running with args: %v\n", args)
  return nil
}

func init() {
  gosubscriber.Subscribe("example_application", "example_queue", testSubscriber, map[string]string{
    "bus_event_type": "testEventOne",
  })
}

func main() {
  if err := gosubscriber.Work(); err != nil {
    fmt.Println("Error:", err)
  }
}

Here is a slightly more complex subscriber that subscribes to the event testEventTwo and prints its arguments, but this example requires the argument required to be present:

package main

import (
  "fmt"
  "github.com/delectable/gosubscriber"
)

func testSubscriber(args map[string]interface{}) error {
  fmt.Printf("testSubscriber running with args: %v\n", args)
  return nil
}

func init() {
  gosubscriber.Subscribe("example_application", "example_queue", testSubscriber, map[string]string{
    "required":       gosubscriber.SpecialValues.Present,
    "bus_event_type": "testEventTwo",
  })
}

func main() {
  if err := gosubscriber.Work(); err != nil {
    fmt.Println("Error:", err)
  }
}

gosubscriber subscribers receive the arguments sent over ResqueBus as a single map of interfaces. To use them as parameters to other functions, use Go type assertions to convert them into usable types.

// where doSomething expects (int64, string)
func testSubscriber(args map[string]interface{}) error {
  idNum, ok := args["id"].(json.Number)
  if !ok {
    return errorInvalidParam
  }
  id, err := idNum.Int64()
  if err != nil {
    return errorInvalidParam
  }

  name, ok := args["name"].(string)
  if !ok {
    return errorInvalidParam
  }
  doSomething(id, name)
  return nil
}

For testing, it's helpful to use IRB to publish events (note that a ResqueBus Driver must be running to process the event)

ResqueBus.publish(:testEventOne)
ResqueBus.publish(:testEventTwo, {required: 1})

For information on configuration/flags, signal handling, and failure modes, see the goworker readme