ihid / propono Goto Github PK
View Code? Open in Web Editor NEWEasy-to-use pub/sub built on AWS. Ruby.
Home Page: http://ihid.github.io/propono/
License: MIT License
Easy-to-use pub/sub built on AWS. Ruby.
Home Page: http://ihid.github.io/propono/
License: MIT License
The error in question is:
Bundler could not find compatible versions for gem "bundler":
In Gemfile:
bundler (~> 1.3)
Options to fix are:
gem
command for thisgemspec
.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!
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
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!
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
This is a nice introduction: https://www.youtube.com/watch?v=ZM3-Gl5DVgM
Add a link somewhere in the README or a Wiki page
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.
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.
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:
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
Suggested Solution:
Give the main queue a 20 second head-start, before listening on both queues. Reset this head-start after ever job processed.
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 💟 !
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?
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
Should the SQS name be a config variable or passed in? I don't rightly know! :)
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.
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.
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?
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
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?
Hi all,
I was wondering if and how I would use propono together with a JS browser client.
Scenario:
Would that work at all? What would be the recommended approach?
Thanks
Christian
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?
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
Line 13 in d0c3685
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'
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
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:
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.
Amazon SQS supports FIFO 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...
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.
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
.
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.
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?
A declarative, efficient, and flexible JavaScript library for building user interfaces.
🖖 Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
An Open Source Machine Learning Framework for Everyone
The Web framework for perfectionists with deadlines.
A PHP framework for web artisans
Bring data to life with SVG, Canvas and HTML. 📊📈🎉
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
Some thing interesting about web. New door for the world.
A server is a program made to process requests and deliver data to clients.
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
Some thing interesting about visualization, use data art
Some thing interesting about game, make everyone happy.
We are working to build community through open source technology. NB: members must have two-factor auth.
Open source projects and samples from Microsoft.
Google ❤️ Open Source for everyone.
Alibaba Open Source for everyone
Data-Driven Documents codes.
China tencent open source team.