Hi,
we are trying to use this plugin to bind from topic exchange to x-history-exchange and the code bombs.
Knowing nothing about erlang, I'm at a bit off a loss but looking around this line it appears that you are only allowing bindings to queues, and not from exchanges (though I'm probably reading this all wrong):

https://github.com/videlalvaro/rabbitmq-recent-history-exchange/blob/master/src/rabbit_exchange_type_recent_history.erl#L75

Is this plugin supposed to bind only to queues? Are we doing something wrong?

Cheers

Publishing to a remote x-recent-history exchange removes the routing key of messages :(

I've recently tried to install that binary and it fails, rabbitmq says:

=ERROR REPORT==== 9-Apr-2015::11:05:13 ===
Problem reading some plugins: [{"/usr/lib/rabbitmq/lib/rabbitmq_server-3.2.4/sbin/../plugins/rabbitmq_recent_history_exchange-1.0.1-rmq3.2.4.ez",
{invalid_ez,einval}}]
1.0.0 is also broken

$ uname -a
Linux ip-172-30-1-180 3.13.0-44-generic #73-Ubuntu SMP Tue Dec 16 00:22:43 UTC 2014 x86_64 x86_64 x86_64 GNU/Linux

$ cat /etc/os-release
NAME="Ubuntu"
VERSION="14.04.2 LTS, Trusty Tahr"
ID=ubuntu
ID_LIKE=debian
PRETTY_NAME="Ubuntu 14.04.2 LTS"
VERSION_ID="14.04"
HOME_URL="http://www.ubuntu.com/"
SUPPORT_URL="http://help.ubuntu.com/"
BUG_REPORT_URL="http://bugs.launchpad.net/ubuntu/"

$ dpkg -s rabbitmq-server
Package: rabbitmq-server
Status: install ok installed
Priority: extra
Section: net
Installed-Size: 4753
Maintainer: Ubuntu Developers [email protected]
Architecture: all
Version: 3.2.4-1
Depends: erlang-nox (>= 1:13.b.3) | esl-erlang, adduser, logrotate
Conffiles:
/etc/logrotate.d/rabbitmq-server ddaa633357e5f6ff9d7970c6d0a9acd8
/etc/default/rabbitmq-server 7cc5216f193f6be1cb32c45021197b39
/etc/init.d/rabbitmq-server 5d49d9839503d8912b13dc1123498257
Description: AMQP server written in Erlang
RabbitMQ is an implementation of AMQP, the emerging standard for high
performance enterprise messaging. The RabbitMQ server is a robust and
scalable implementation of an AMQP broker.
Original-Maintainer: RabbitMQ Team [email protected]
Homepage: http://www.rabbitmq.com/

Please provide such doc, I think for many folks it's critical factor when including in production setup.
BTW I think that your project is a lot more than just chat sync :), I like it so much.

Line 121, Msgs list should be reversed before forwarded through destination exchange

This might 1) be by design or 2) because I'm doing something wrong (probably the case...), but I'm only seeing the message history delivered to new queues.

Here's my code (I'm using the Bunny Ruby client):

@conn = Bunny.new(ENV['CLOUDAMQP_URL'])
@conn.start
@ch = @conn.create_channel
@x = @ch.exchange("video-exchange",
                  type: "x-recent-history",
                  arguments: { 'x-recent-history-length' => 1 })

# will only receive history payload once
queue_name = "queue-1"
@ch.queue(queue_name).bind(@x).subscribe do |_, _, payload|
  puts "!!!!!!!!! '#{payload}' on #{queue_name}"
end

# will always receive history payload
queue_name = "queue-#{SecureRandom.hex}"
@ch.queue(queue_name).bind(@x).subscribe do |_, _, payload|
  puts "!!!!!!!!! '#{payload}' on #{queue_name}"
end

Hi

We wanted to use this plugin in a HA environment, however I'm running into a bit of a problem
with the x-recent-history exchanges with the following setup;

• 2 nodes setup in a cluster ( rabbitmq 3.3.5 )
• A policy set with the parameters "ha-mode: all"

This replicates the queues and exchanges across both machines.

When both nodes are up, we can bind a queue to one of the recent history exchanges from either of the nodes.

When Node 1 where the x-recent-history exchange was created is shut down, then trying to bind a queue to the exchange on Node 2 gives the following error:

** Reason for termination ==
** {{error,{throw,{error,{no_exists,rh_exchange_table}}}},
[{rabbit_misc,execute_mnesia_transaction,1},
{rabbit_misc,execute_mnesia_tx_with_tail,1},
{rabbit_channel,binding_action,9},
{rabbit_channel,handle_cast,2},
{gen_server2,handle_msg,2},
{proc_lib,init_p_do_apply,3}]}

Unfortunately my erlang is not the best, however from what I can see it appears the in memory table that is setup for the exchange which holds the history events is only created on the node that made the exchange. So when that node is shut down the replica cannot access the table and fails.

Are there any special requirements for creating the exchanges on clustered nodes or is there a way for this table to replicate between nodes?

Thanks
Oliver

Hi,

This Plugin is looks very nice. Please provide me binary for RabbitMQ 3.4.1.

Thanks
memorycraft

As mentioned in the the title, it would be nice to be able to change the history/replay size in the RabbitMQ configuration file.

This plugin would be more useful if it also supported a method of storing (and redelivering) messages younger than N seconds.

My use case involves clients connecting to exclusive queues to receive data updates. When a client connects, its data may be up to a minute out of date. Limiting the cache to a fixed number of messages doesn't work well when the frequency at which messages are pushed is highly variable. If the message limit is small, pushing many messages in a short period of time may cause messages to not be delivered to new queues. On the other hand, if the message limit is large, there will be messages in the queue that can be hours old that the client will need to handle.

Is anyone using this in production?
Whats the status of the plugin, was it only created for the chat example?

As part of our across the board move to Common Test.

Has anybody knowledges of mirroring x-recent-exchage in cluster?

After some experiments with restart rabbitMq I've observed no recovered messages in historical exchange. So it would be better experience if persistance of messages in historical exchange will appear.

hi, I'm trying to get your plugin working on rabbitmq 3.0.4 (erlang R16B), but unfortunately every time the queue binds to exchange of "x-recent-history" type, following error is thrown:

01:10:57.731 [error] Connection (<0.76.0>) closing: received hard error {'connection.close',541,<<"INTERNAL_ERROR">>,0,0} from server

plugin itself seem to be recognized and exchange is correctly created:

-- plugins running
rabbitmq_recent_history_exchange  0.1.0-rmq0.0.0

Any hints what may be the cause of this problem?

By the way, how does this plugin route incoming messages? Does it work like a fanout?
Thanks,
Michał

Setting the queue with 'x-recent-history-length: 1000000'

args={'x-recent-history-length':1000000}
channel.exchange_declare(exchange='buffer', exchange_type='x-recent-history', arguments =args)

for i in range(1000000):
    message = datetime.now().strftime("%H:%M:%S") + '.' + str(i)
    channel.basic_publish(exchange='buffer',
                          routing_key='',
                          body=message)

At first it publish 10000 per second. after 100000 msgs , it starts to slow down , and then after 200000, it broke.

Hi,

Thanks for making this plugin, it looks like it could be handy for a usecase I need to address. I'm trying to set up the following RabbitMQ pub/sub system:

• Publishers write messages with various topics to a main exchange of type topic.
• The main exchange is bound to a second exchange, of type x-recent-history. All messages are routed between the two (routing_key == '#').
• The subscribers connect to the second exchange.

I would like that upon connection, the subscribers only receive the last N messages related to their topic. Right now they receive N messages with any topic. I see that the topic argument is ignored for the recent history queue. I'm wondering if this is by design or if it's a bug?

Thanks!
Radu

I've installed and enabled this plugin, but still can't figure how to use it. Could you add some basic information about using this plugin?

What will happen on a restart, will the last X messages be available?
Does this affect replication between a cluster as well, is it available across the cluster os just on one node?

publisher publish 1000000 msg. consumer starts later to receive the msg. it hangs after 100000 msgs. Code shown below:

Publisher

args={'x-recent-history-length':1000000}
channel.exchange_declare(exchange='buffer', exchange_type='x-recent-history', arguments =args)
for i in range(1000000):
    message = datetime.now().strftime("%H:%M:%S") + '.' + str(i)
    channel.basic_publish(exchange='buffer',
                          routing_key='',
                          body=message)

Consumer

channel.exchange_declare(exchange='buffer', exchange_type='x-recent-history')
result = channel.queue_declare(exclusive=True)
queue_name = result.method.queue
channel.queue_bind(exchange='buffer',
                   queue=queue_name)
print(' [*] Waiting for buffer. To exit press CTRL+C')

def callback(ch, method, properties, body):
    msg_id = int (body.split('.')[1])
    if msg_id % 10000 == 0:
        print(" [x] %s:  %r" % (datetime.now().strftime("%H:%M:%S"),body))

Note: this might be related to issue #27