"KRAKOW! KRAKOW! Two direct hits!"
require 'krakow'
producer = Krakow::Producer.new(
:host => 'HOST',
:port => 'PORT',
:topic => 'target'
)
producer.write('KRAKOW!', 'KRAKOW!')require 'krakow'
consumer = Krakow::Consumer.new(
:nsqlookupd => 'http://HOST:PORT',
:topic => 'target',
:channel => 'ship'
)
consumer.queue.size # => 2
2.times do
msg = consumer.queue.pop
puts "Received: #{msg}"
consumer.confirm(msg.message_id)
endIt's a Ruby library for NSQ using Celluloid under the hood.
Yep, that's right. Just one lowly message at a time. And that's probably not what you want, so adjust it when you create your consumer instance.
require 'krakow'
consumer = Krakow::Consumer.new(
:nsqlookupd => 'http://HOST:PORT',
:topic => 'target',
:channel => 'ship',
:max_in_flight => 30
)Since Celluloid is in use under the hood, and the main interaction points are
Actors (Consumer and Producer) you'll need to be sure you clean up. This simply
means terminating the instance (since falling out of scope will not cause it to be
garbage collected).
consumer = Krakow::Consumer.new(
:nsqlookupd => 'http://HOST:PORT',
:topic => 'target',
:channel => 'ship',
:max_in_flight => 30
)
# do stuff
consumer.terminateSure:
Krakow::Utils::Logging.level = :warn # :debug / :info / :warn / :error / :fatalBecause forcing starvation is mean. We don't want to be mean, so we'll ensure we are consuming from all registered connections.
Fine!
consumer = Krakow::Consumer.new(
:host => 'HOST',
:port => 'PORT',
:topic => 'target',
:channel => 'ship',
:max_in_flight => 30
)Great for testing, but you really should use the lookup service in the "real world"
NSQ has this backoff notion. It's pretty swell. Basically, if messages from a specific producer get re-queued (fail), then message consumption from that producer is halted, and slowly ramped back up. It gives time for downstream issues to work themselves out, if possible, instead of just keeping the firehose of gasoline on. Neat.
By default backoff support is disabled. It can be enabled by setting the :backoff_interval
when constructing the Consumer. The interval is in seconds (and yes, floats are allowed
for sub-second intervals):
consumer = Krakow::Consumer.new(
:nsqlookupd => 'http://HOST:PORT',
:topic => 'target',
:channel => 'ship',
:max_in_flight => 30,
:backoff_interval => 1
)OK!
consumer = Krakow::Consumer.new(
:nsqlookupd => 'http://HOST:PORT',
:topic => 'target',
:channel => 'ship',
:connection_options => {
:features => {
:tls_v1 => true
}
}
)OK!
consumer = Krakow::Consumer.new(
:nsqlookupd => 'http://HOST:PORT',
:topic => 'target',
:channel => 'ship',
:connection_options => {
:features => {
:snappy => true
}
}
)NOTE: snappy support requires the snappy gem and is not provided by default, so you will need to ensure it is installed either on the system, or within the bundle.
OK!
consumer = Krakow::Consumer.new(
:nsqlookupd => 'http://HOST:PORT',
:topic => 'target',
:channel => 'ship',
:connection_options => {
:features => {
:deflate => true
}
}
)OK!
consumer = Krakow::Consumer.new(
:nsqlookupd => 'http://HOST:PORT',
:topic => 'target',
:channel => 'ship',
:connection_options => {
:features => {
:tls_v1 => true
},
:config => {
:ssl_context => {
:certificate => '/path/to/cert',
:key => '/path/to/key'
}
}
}
)Run them all!
bundle exec ruby test/run.rb
Or, run part of them:
bundle exec ruby test/specs/consumer_spec.rb
NOTE: the specs expect that nsqd and nsqlookupd are available in $PATH
Create an issue on the github repository
Create an issue, or even better, send a PR.
- Repo: https://github.com/chrisroberts/krakow
- Docs: http://code.chrisroberts.org/krakow
- IRC: Freenode @ spox
- Pete Hopkins (@phopkins)
- Sam Phillips (@i2amsam)
- Brendan Schwartz (@bschwartz)
- Thomas Holmes (@thomas-holmes)
- Jeremy Hinegardner (@copiousfreetime)
React
Typescript
Django
Laravel
Facebook
Microsoft
Google
Alibaba
D3
Tencent