Pulsar

Pulsar is a simple Crystal library for publishing and subscribing to events. It also has timing information for metrics. So what does that mean in practice?

You can define an event and any number of subscribers can subscribe to the event and do whatever they need with it.

For example, in Lucky, we use Pulsar to create events for things like requests being processed, queries being made, before and after pipes running. Then we subscribe to these events to write to the logs. We also use this internally to log debugging information in an upcoming UI called Breeze that let's users debug development information.

Installation

Add the dependency to your shard.yml:

dependencies:
  pulsar:
    github: luckyframework/pulsar

Run shards install

How to use Pulsar

Let's say we're writing a library to charge a credit card and we may want to let people run code whenever a charge is made. Here's how you can do that with Pulsar.

Create and publish an event

class PaymentProcessor::ChargeCardEvent < Pulsar::Event
  def initialize(@amount : Int32)
  end
end

class PaymentProcessor
  def charge_card(amount : Int32)
    # Run code to charge the card...

    # Then fire an event
    PaymentProcessor::ChargeCardEvent.publish(amount)
  end
end

Subscribe to it and do whatever you want with it

Now you can subscribe to the event and do whatever you want with it. For example, you might log that a charge was made, or you might send an email to the sales team.

PaymentProcessor::ChargeCardEvent.subscribe do |event|
  puts "Charged: #{event.amount} at #{event.started_at}"
end

Recording timing information

You can also time how long it takes to run an event by inheriting from Pulsar::TimedEvent. You define them in the same way, but when you subscribe you must also accept a second argument:

class Database::QueryEvent < Pulsar::TimedEvent
end

Database::QueryEvent.subscribe do |event, duration|
  # Do something with the event and duration
end

Database::QueryEvent.publish do
  # Run a query, run some other code, etc.
end

Add more information to the event

To add more information to the event you can use initialize like you would with any other Crystal class.

For example, we can record the database query in the event from above

class Database::QueryEvent < Pulsar::TimedEvent
  getter :query

  def initialize(@query : String)
  end
end

Database::QueryEvent.subscribe do |event, duration|
  puts event.query
end

Database::QueryEvent.publish(query: "SELECT * FROM users") do
  # Run a query, run some other code, etc.
end

Testing Pulsar events

If you want to test that events are published you can use Pulsar's built-in test mode.

# Typically in spec/spec_helper.cr

# Must come *after* `require "spec"`
Pulsar.enable_test_mode!

This will enable an in-memory log for published events and will set up a hook to clear the events before each spec runs.

You can access events using {MyEventClass}.logged_events.

# Create an event
class QueryEvent < Pulsar::TimedEvent
  def initialize(@query : String)
  end
end

def run_my_query(query)
  # Publish the event
  QueryEvent.publish(query: query) do
    # Run the query somehow
  end
end

it "publishes an event when a SQL query is executed" do
  run_my_query "SELECT * FROM users

  the_published_event = QueryEvent.logged_events.first
  the_published_event.query.should eq("SELECT * FROM users")
end

`Pulsar.elapsed_text`

Pulsar.elapsed_text will return the time taken (Time::Span) as a human readable String.

Database::QueryEvent.subscribe do |event, duration|
  puts Pulsar.elaspted_text(duration) # "2.3ms"
end

This method can be used with any Time::Span.

Performance gotchas

Subscribers are notified synchronously in the same Fiber as the publisher. This means that if you have a subscriber that takes a long time to run, it will block anything else from running.

If you are doing some logging it is probably fine, but if you are doing something more time-intensive or failure prone like making an HTTP request or saving to the database you should pay special attention.

Example of a problematic subscriber

MyEvent.subscribe do |event|
  sleep(5)
end

MyEvent.publish

puts "I just took 5 seconds to print!"

Oops. To get around this you can spawn a new fiber:

MyEvent.subscribe do |event|
  # Now the `sleep` will run in a new Fiber and will not block this one
  spawn do
    sleep(5)
  end
end

MyEvent.publish

puts "This will print right away!"

Potential solutions

As described above you could run long running code in a new Fiber with spawn. You could also use a background job library like https://github.com/robacarp/mosquito.

Be aware that running things in a Fiber will lose the current Fiber's context. This is important for logging since Log.context only works for the current Fiber. So if you plan to log using the built-in Logger, you likely do not want to spawn a new fiber. It is fast enough to just log like normal.

Contributing

Fork it (https://github.com/luckyframework/pulsar/fork)
Create your feature branch (git checkout -b my-new-feature)
Commit your changes (git commit -am 'Add some feature')
Push to the branch (git push origin my-new-feature)
Create a new Pull Request

Contributors

paulcsmith - creator and maintainer

CrystalDoc.info

pulsar