ihid / propono Goto Github PK

View Code? Open in Web Editor NEW

317.0 317.0 27.0 451 KB

Easy-to-use pub/sub built on AWS. Ruby.

Home Page: http://ihid.github.io/propono/

License: MIT License

Ruby 100.00%

propono's Introduction

Hi there 👋

I'm Jeremy (he/him) - a social entrepeneur and software developer (👨‍💻). Online I go by iHiD.

You probably know me as the co-founder of Exercism - an online platform that helps people upskill their programming skills through practice and mentoring. We're a open-source organisation with over 350 GitHub repositories, thousands of contributors, and a friendly, inclusive community. We've had over 1.8M people use us and millions of submssions. We're always looking for new people to help - email me if you'd like to get involved 🙂

You might also know me as the co-founder of Kaido - a health and wellbeing startup helping companies take better care of their employees. We work with companies who are tiny (4 people), huge multinationals (e.g. HSBC and KPMG), charities (e.g. Refugee Action) and with the NHS - the UK's public health service. Try Kaido for free and see how we can help your workplace be a happier and healthier place! 🙂

I also run Thalamus. I previously co-founded Wondr Medical (2016-2019) and Meducation (2008-2016). I've was also a founding member of Flaunt and GetStoic, and ran two software consultancies (Meridian Innovations and Pure Intellect).

From 2016-2019 I was a Non-exec Director at the wonderful Beatfreeks.

Bio

I spend my time building organisations and businesses that help others. My core skillset involves strategy, leadership, community building, and software development. I have a degree in Artificial Intelligence and have co-developed a deep-learning algorithm used in Interventional Cardiology.

Day-to-day, I'm normally coding, leading, strategising or thinking - whichever is the most important at any given time!

Away from work I read, travel, climb, and get nerdy about coffee.

propono's People

Contributors

Stargazers

Watchers

propono's Issues

sending from a rake task never arrives

I just created simple 'publish' and 'listen' rake tasks. While they work from the console, when I move them into rake tasks, the published messages never arrive at the listener. I added a 'sleep 60' at the end of the rake task, and then I start receiving the messages. Apparently the termination of ruby when the rake task ends somehow halts the message. Is there a thread I should join on?

After receiving an error reading from an sqs queue, propono spins and never recovers.

The ouput below is taken for the convorto daemon logs. It shows the convorto listener thread receiving an unexpected 403 Forbidden response. After recieving the error, it continuously loged the same error, multiple times a second until the process was eventually stopped.

2013-12-07 04:09:58 UTC ip-10-73-171-116 convorto 2013-12-07T04:10:23+00:00 thread_id-20190640 [] INFO | Signed request executed. Response: [] | /opt/convorto/vendor/bundle/ruby/2.0.0/gems/loquor-0.5.1/lib/loquor/http_actions/get.rb:14:in `get'
2013-12-07 04:09:58 UTC ip-10-73-171-116 convorto 2013-12-07T04:10:23+00:00 thread_id-20190640 [] INFO | Poll complete. | /opt/convorto/lib/convorto/pollers/wistia_poller.rb:40:in `poll_all_processing'
2013-12-07 04:10:28 UTC ip-10-73-171-116 convorto 2013-12-07T04:10:56+00:00 thread_id-20189040 [] INFO | There are 2 worker threads running | /opt/convorto/lib/tasks/convorto.rake:82:in `block (3 levels) in <top (required)>'
2013-12-07 04:10:48 UTC ip-10-73-171-116 convorto 2013-12-07T04:11:12+00:00 thread_id-20191320 [94e117] ERROR | Unexpected error reading from queue https://sqs.eu-west-1.amazonaws.com:443/950417255687/convorto-media_file | /opt/convorto/vendor/bundle/ruby/2.0.0/gems/propono-0.11.0/lib/propono/services/queue_listener.rb:35:in `rescue in read_messages'
2013-12-07 04:10:48 UTC ip-10-73-171-116 convorto 2013-12-07T04:11:12+00:00 thread_id-20191320 [94e117] ERROR | Expected(200) <=> Actual(403 Forbidden)
2013-12-07 04:10:48 UTC ip-10-73-171-116 convorto  response => #<Excon::Response:0x000000026a5f98 @data={:body=>"<?xml version=\"1.0\"?><ErrorResponse xmlns=\"http://queue.amazonaws.com/doc/2009-02-01/\"><Error><Type>Sender</Type><Code>ExpiredToken</Code><Message>The security token included in the request is expired</Message><Detail/></Error><RequestId>765ec6bd-4fce-5da0-b838-8da1edc1c6b0</RequestId></ErrorResponse>", :headers=>{"Server"=>"Server", "Date"=>"Sat, 07 Dec 2013 04:10:55 GMT", "Content-Type"=>"text/xml", "Content-Length"=>"302", "Connection"=>"keep-alive", "x-amzn-RequestId"=>"765ec6bd-4fce-5da0-b838-8da1edc1c6b0"}, :status=>403, :remote_ip=>"176.32.111.157"}, @body="<?xml version=\"1.0\"?><ErrorResponse xmlns=\"http://queue.amazonaws.com/doc/2009-02-01/\"><Error><Type>Sender</Type><Code>ExpiredToken</Code><Message>The security token included in the request is expired</Message><Detail/></Error><RequestId>765ec6bd-4fce-5da0-b838-8da1edc1c6b0</RequestId></ErrorResponse>", @headers={"Server"=>"Server", "Date"=>"Sat, 07 Dec 2013 04:10:55 GMT", "Content-Type"=>"text/xml", "Content-Length"=>"302", "Connection"=>"keep-alive", "x-amzn-RequestId"=>"765ec6bd-4fce-5da0-b838-8da1edc1c6b0"}, @status=403, @remote_ip="176.32.111.157"> | /opt/convorto/vendor/bundle/ruby/2.0.0/gems/propono-0.11.0/lib/propono/services/queue_listener.rb:36:in `rescue in read_messages'
2013-12-07 04:10:48 UTC ip-10-73-171-116 convorto 2013-12-07T04:11:12+00:00 thread_id-20191320 [94e117] ERROR | Unexpected error reading from queue https://sqs.eu-west-1.amazonaws.com:443/950417255687/convorto-media_file | /opt/convorto/vendor/bundle/ruby/2.0.0/gems/propono-0.11.0/lib/propono/services/queue_listener.rb:35:in `rescue in read_messages'
2013-12-07 04:10:48 UTC ip-10-73-171-116 convorto 2013-12-07T04:11:12+00:00 thread_id-20191320 [94e117] ERROR | Expected(200) <=> Actual(403 Forbidden)
2013-12-07 04:10:48 UTC ip-10-73-171-116 convorto  response => #<Excon::Response:0x000000027498c8 @data={:body=>"<?xml version=\"1.0\"?><ErrorResponse xmlns=\"http://queue.amazonaws.com/doc/2009-02-01/\"><Error><Type>Sender</Type><Code>ExpiredToken</Code><Message>The security token included in the request is expired</Message><Detail/></Error><RequestId>c13dd968-f8dc-5dad-83ea-92745ad3e625</RequestId></ErrorResponse>", :headers=>{"Server"=>"Server", "Date"=>"Sat, 07 Dec 2013 04:10:55 GMT", "Content-Type"=>"text/xml", "Content-Length"=>"302", "Connection"=>"keep-alive", "x-amzn-RequestId"=>"c13dd968-f8dc-5dad-83ea-92745ad3e625"}, :status=>403, :remote_ip=>"176.32.111.157"}, @body="<?xml version=\"1.0\"?><ErrorResponse xmlns=\"http://queue.amazonaws.com/doc/2009-02-01/\"><Error><Type>Sender</Type><Code>ExpiredToken</Code><Message>The security token included in the request is expired</Message><Detail/></Error><RequestId>c13dd968-f8dc-5dad-83ea-92745ad3e625</RequestId></ErrorResponse>", @headers={"Server"=>"Server", "Date"=>"Sat, 07 Dec 2013 04:10:55 GMT", "Content-Type"=>"text/xml", "Content-Length"=>"302", "Connection"=>"keep-alive", "x-amzn-RequestId"=>"c13dd968-f8dc-5dad-83ea-92745ad3e625"}, @status=403, @remote_ip="176.32.111.157"> | /opt/convorto/vendor/bundle/ruby/2.0.0/gems/propono-0.11.0/lib/propono/services/queue_listener.rb:36:in `rescue in read_messages'
2

Queue size and oldest message?

To put Propono in the environment I’d like with the monitoring I need, I’d like to be able to get the current size of the queue as well as the age of the oldest message. I don’t see that in the current api... I can look at adding it, but I was wondering if you thought there was a technical limitation why these weren’t there yet.

Use IAM role instead of access & secret keys

I am attempting to refactor an application which has been successfully using this gem. Currently it uses AWS access & secret keys in the propono config, however when refactoring it to use an IAM role, it currently throws an error at the following line:

https://github.com/iHiD/propono/blob/master/lib/propono/configuration.rb#L38

I believe this is just because access_key and secret_key are not being initialized, so unless you specifically set those values, an error will be thrown.

wrong number of arguments (given 5, expected 4)

I am getting an Argument Error when using trying to publish a message. I am using propono 3.0, ruby-3 and rails-6

[GEM_ROOT]/gems/propono-3.0.0/lib/propono/services/publisher.rb:14:in initialize' initialize: wrong number of arguments (given 5, expected 4) (ArgumentError)
[GEM_ROOT]/gems/propono-3.0.0/lib/propono/services/publisher.rb:9:in new'
[GEM_ROOT]/gems/propono-3.0.0/lib/propono/services/publisher.rb:9:in publish'
[GEM_ROOT]/gems/propono-3.0.0/lib/propono/components/client.rb:54:in publish'

Are we having the support of Ruby3?

vs Hutch?

How would you compare propono against Hutch?

I think Rails app really needs nice & easy solution like propono & hutch that work out of the box for inter-app communication so keep up the good work 💟 !

Slow queue wait implementation causes delay in processing of new main queue messages

If the main queue is empty after 20 seconds, the client then listens on the slow queue for 20 seconds. While listening on the slow queue, any item arriving in the main queue will have to wait till the slow queue listen is over, and the next main queue listen, to be processed.

Timeline:

Listener starts to wait on main queue
Nothing to process
Listener stops listening on main queue after 20 seconds
Listener starts to wait on slow queue
Nothing to process
After 1 second an item is added to the main queue
Listener stops listening on the slow queue after a further 19 seconds
Listener starts to wait on main queue
Item in main queue is immediately received and processed, 19 seconds after it was published.

Code to show in action:

topic  = :slow_queue_delay_test
client = Propono::Client.new

Thread.new do
  client.listen(topic) do |payload|
    puts "Message processed after #{Time.now.to_i - payload[:time]} seconds"
  end
end

# One second longer than the long-poll time.
sleep 21
PublishProponoMessage.(topic, { time: Time.now.to_i })

# Keep the main thread alive.
sleep 30

Update repository

Hi, I'm seeing that a new version of the gem has been released (https://rubygems.org/gems/propono/versions/2.1.0) but I can't see any changes published to this repo.
Is this still the main page for the project or has it been changed?
Probably update

propono/propono.gemspec

Line 13 in d0c3685

spec.homepage = ""

to reflect this changes
or update the repository with the latest revision and tags

Way to use FIFO queues?

Amazon SQS supports FIFO queues:

https://aws.amazon.com/about-aws/whats-new/2016/11/amazon-sqs-introduces-fifo-queues-with-exactly-once-processing-and-lower-prices-for-standard-queues/

Any consideration given to supporting this?

My initial investigation of the code makes it seem like it would just require an update to the Fog-AWS dependency and some new options passed at queue creation time... but I'm not sure what I'd want the semantics of the setup to look like... is this something set on Propono.config before we publish to the queue? The actual creation of the queue is hidden from the usage of the api...

Javascript subscriber

Hi all,

I was wondering if and how I would use propono together with a JS browser client.

Scenario:

Sinatra app which uses propono to publish a json message to SNS/SQS.
Browser (JS) subscribes to the SQS queue to receive the message from Sinatra. As far as I know the browser would only be able to poll the SQS queue, so no real pushing of messages to the browser.

Would that work at all? What would be the recommended approach?

Thanks
Christian

Travis CI Build Failures Due to Bundler Version

Example failing build

The error in question is:

Bundler could not find compatible versions for gem "bundler":
  In Gemfile:
    bundler (~> 1.3)

Options to fix are:

Fix the bundler version travis uses to 1.X. gem command for this
Upgrade the bundler version in the gemspec.

How do I use GoAWS?

https://github.com/p4tin/goaws

I am trying to change the endpoint for AWS to use local environment for SQS and SNS, but I couldn't find the right place to customize it. Any idea where I can use the local URL?

Thanks,
Praveen

Mocking SMS/SQS

I was curious if you had any strategies for using applications without AWS access. Perhaps we could mock sms/sqs with a simple but functional esb in the backend. I realized there exists issue #45 which discusses simple mocking of propono methods for testing, but I am more curious about setting up application environments for user and development testing.

Has anyone ended up trying out ActiveJob or similar for the purpose?

testing strategies?

This is likely addressed with a section in the readme, but what are some guidelines for testing without relying on a live SQS instance? Do you mock the Propano.publish and .listen_to_queue methods and check for the messages you expect?

Docs - Flowchart of corrupted/failed/etc queue logic

Add link to your talk

This is a nice introduction: https://www.youtube.com/watch?v=ZM3-Gl5DVgM

Add a link somewhere in the README or a Wiki page

Message Visibility Timeout is too short for some classes of tasks

Failure case

By default the timeout visibility of a message in an SQS queue is 30 seconds. If processing a video, say, which takes an hour to process, the worker only has exclusivity on the message for the first 30s of the processing time. After that (depending on the queue setup, I haven't looked closely at it), the message is either dequeued for other workers to pick up, or put in a dead letter queue.

This can result in multiple workers processing the same message, or a single worker processing the same message repeatedly.

Suggested Solution

Aws::SDK::SQS#receive_message allows setting of a visibility_timeout attribute (0s to 12h, defined in seconds). Thus it would seem that setting visibility_seconds on the listen method would prevent this issue.

Propono::Client.new.listen('long-running-tasks', visibility_timeout: 3600) do |message|
  puts "I just received: #{message}"
end

Perhaps a better, and less AWS specific, key might be something like exclusivity_timeout_seconds.

SQS Long Polling?

Firstly, this looks great! Been leafing through the source code and its nice to see a well structured, neatly organised collection of concerns.

The objective of leafing through was to see if you had chosen to use SQS Long Polling which it appears that you havent. Was there a reason behind this decision?

Need a way to pass Subject for the SNS payload

Hey Jeremy,

Apologies if this is the wrong place to put this, but I wasn't sure where to make the request.

I am using Propono to publish SNS messages, and then consuming them with Lambdas that integrate with various 3rd party services (like Slack, and many others).

Our topics are notifiable elements (essentially business objects), with one Lambda acting as a message broker, figuring out which integration to hit, via resubmission of a SNS package against a modified topic. The modified topic is the original topic plus the integration (i.e. user-slack).

The "Slack" Lambda is triggered from that message and fires of a message to Slack.

My issue is that I need the original path accessible to the broker Lambda, and while I could parse the ARN, it would be much easier to just use the Subject of the SNS package, which seems to always be nil with Propono-generated messages.

I altered the services/publisher.rb script by passing options to the sns.publish method:

def publish_via_sns_syncronously
      begin
        topic = TopicCreator.find_or_create(topic_id)
      rescue => e
        Propono.config.logger.error "Propono [#{id}]: Failed to create topic #{topic_id}: #{e}"
        raise
      end

      begin
        sns.publish(topic.arn, body.to_json, {'Subject' => topic_id})
      rescue => e
        Propono.config.logger.error "Propono [#{id}]: Failed to send via sns: #{e}"
        raise
      end
end

While this works, it would, obviously, be better if it were actually parameterized so I could pass it in with the call to Propono.publish and not have to mod your code.

Is there any issue you see with this? If not, I can put it together in a pull and send it up.

Cheers,

Chris

Read from failed queue?

What's the best way to read from the failed queue? I can see how messages get added to the failed queue, but what's the best way to read/drain the failed queue later on (maybe a bug has been fixed and ready to re-read the messages again...)

Thanks!

Docs - Note about needing a subscriber before you publish

Asynchronous listen_to_queue

I appreciate asynchronous publishing, but that makes subscribing a bottleneck. I seem to reach subscribing speeds of around 0.6-1 message/second using the Propono.listen_to_queue method with one thread.

I later tested this with multiple threads and that does speed it up. It would be convenient if this were built into Propono. I imagine that long polling in v2 should help with this.

  def async_subscribe(i)
    i.times.each do |ii|
      puts "created thread #{ii}"
      @threads << Thread.new(messages) { subscribe }
    end
    @threads.join
  end

  def subscribe
    Propono.listen_to_queue(:foobar) do |message|
      puts message
    end
  end

Should we merge iterum into Propono

So the only thing now blocking me from releasing the pre-release of v2 is reading from failed queues. We seem to have once had code that to have code for this but it got refactored out at some stage. It's now hardcoded via iterum (which will stop working). Do we want to bake this into Propono core somehow.

@malcyL @ccare Thoughts pls.

Propono should error when starting to listen to nil topic

udp_listener failing to parse JSON

There are two issues here.

The call to "json = JSON.parse(udp_data)" needs a "require 'json'"
After adding in the "require 'json'" we are not able to parse the messages sent by either the meducation app or the udp2sqs_client/bin/send.rb test.

propono is now expecting a 'topic' and 'message' in the root of the json it receives over udp. Both the send.rb and meducation are sending:

{"action"=>"pageview", "time"=>"", "payload"=>{"session_id"=>"1234", "user_id"=>nil, "premium"=>false}}

We do want to move to a topic and a message, but releasing propono as is will result in the dashboard breaking.

Perhaps propono's UdpListener should handle the old and new message formats. The old format can be removed from propono once meducation is publishing in the new format.

Set the queue name

Should the SQS name be a config variable or passed in? I don't rightly know! :)

License info on gh-pages branch inconsistent with LICENSE.md

The license info at https://meducation.github.io/propono (from the gh-pages branch) refers to the Affero GPLv3, while the LICENSE.md in the master branch and the README.md refer to the MIT license.

Docs - Description/diagram of single-queue-multiple-listener vs fanout

messages being processed multiple times by same worker name

According to the readme, "subscribers that share the same application_name will act as multiple workers on the same queue. Only one will get to process each message", but this isn't true.

I have an app where processing each message is taking ~30 seconds of processor time.

Because of the code here:

propono/lib/propono/services/queue_listener.rb

Line 66 in 7b0dfe8

handle(sqs_message)

while the first worker is processing the message, a second worker has the opportunity to grab it.

Reversing those two lines kinds solves this issue, but creates a new one... a message can get lost if the worker fails during processing.

I don't think the semantics of sqs and sns alone can fix this issue one way or the other; sqs would need to be able to mark a message as 'processing'.

Perhaps deleting the message, calling the message handler, rescuing an exception, republishing the message, then re-raising the exception would be a potential strategy for handling this, but certainly raises the complexity.

Multiple listen blocks on the same file

Hi, what's the right approach to follow if I want to listen to many different topics on the same script?
For the subscriber part of my pub/sub architecture I want to have one script running (as a daemon on the background) that can listens to multiple topics and do something depending on the topic.
I have the following script (inside a Rails app):

#!/usr/bin/env ruby
require File.expand_path('../config/environment.rb',  __FILE__)

# setting up the client
client = Propono::Client.new
client.config.access_key = my_access_key
client.config.secret_key = my_secret_key
client.config.queue_region = region
client.config.application_name = "my-unique-app-name"

# listening to different kind of events so we can respond according
client.listen("topic-1") do |message|
  puts "Topic 1: #{message}"
  # do something related to topic-1
end

client.listen("topic-2") do |message|
  puts "Topic 2: #{message}"
  # do something related to topic-2
end

client.listen("topic-3") do |message|
  puts "Topic 3: #{message}"
  # do something related to topic-3
end

However this is not working and if I run the above ☝️script, it'll only respond to messages/topics happened on topic-1, and I found it's because the listen method is implemented as an endless loop do block, so it never returns.

Is there a way to accomplish the above without having to separate each listen do block into its own script? I don't want to have multiple scripts (one per topic) running on the background individually.

Thanks

ArgumentError: invalid configuration option `:use_iam_profile'

Hello All,

I am receiving the following error when attempting to publish a message. Was use_iam_profile deprecated when FOG was removed?

CALL:

client = Propono::Client.new do |config|
  config.use_iam_profile = true
  config.queue_region = "us-east-1"
  config.application_name = "PermissionServiceBus"
  config.logger = Rails.logger
end

client.publish('topic1', 'A Simple Message')

ERROR:

ArgumentError: invalid configuration option `:use_iam_profile'
from .rvm/gems/ruby-2.3.4@wikiposit/gems/aws-sdk-core-3.21.2/lib/seahorse/client/configuration.rb:166:in `rescue in block in apply_options'
from .rvm/gems/ruby-2.3.4@wikiposit/gems/aws-sdk-core-3.21.2/lib/seahorse/client/configuration.rb:162:in `block in apply_options'
from .rvm/gems/ruby-2.3.4@wikiposit/gems/aws-sdk-core-3.21.2/lib/seahorse/client/configuration.rb:161:in `each'
from .rvm/gems/ruby-2.3.4@wikiposit/gems/aws-sdk-core-3.21.2/lib/seahorse/client/configuration.rb:161:in `apply_options'
from .rvm/gems/ruby-2.3.4@wikiposit/gems/aws-sdk-core-3.21.2/lib/seahorse/client/configuration.rb:149:in `build!'
from .rvm/gems/ruby-2.3.4@wikiposit/gems/aws-sdk-core-3.21.2/lib/seahorse/client/base.rb:62:in `build_config'
from .rvm/gems/ruby-2.3.4@wikiposit/gems/aws-sdk-core-3.21.2/lib/seahorse/client/base.rb:19:in `initialize'
from .rvm/gems/ruby-2.3.4@wikiposit/gems/aws-sdk-sns-1.2.0/lib/aws-sdk-sns/client.rb:143:in `initialize'
from .rvm/gems/ruby-2.3.4@wikiposit/gems/aws-sdk-core-3.21.2/lib/seahorse/client/base.rb:99:in `new'
from .rvm/gems/ruby-2.3.4@wikiposit/gems/propono-2.1.0/lib/propono/components/aws_client.rb:71:in `sns_client'
from .rvm/gems/ruby-2.3.4@wikiposit/gems/propono-2.1.0/lib/propono/components/aws_client.rb:26:in `create_topic'
from .rvm/gems/ruby-2.3.4@wikiposit/gems/propono-2.1.0/lib/propono/services/publisher.rb:41:in `publish_syncronously'
from .rvm/gems/ruby-2.3.4@wikiposit/gems/propono-2.1.0/lib/propono/services/publisher.rb:30:in `publish'
from .rvm/gems/ruby-2.3.4@wikiposit/gems/propono-2.1.0/lib/propono/services/publisher.rb:9:in `publish'

How to use propono for one-to-many broadcast?

I've just started looking at propono and so far love the ease of use.

The specific scenario I'd like to support is publishing message to multiple consumers (without the publisher needing to know if any/how many subscribers are listening).

Out of the box, this does not appear to work, because messages are dequeued as soon as read by one subscriber. i.e. only one of the many subscribers will ever get the message.

But maybe I'm overlooking some obvious configuration/usage guidelines to get 1-many to work.

I'd appreciate it if anyone can advise of this is possible (and if so how) before I get too lost on a yak-shaving expedition..

Cheers!