diff options
Diffstat (limited to 'extensions/Push/lib/Connector')
| -rw-r--r-- | extensions/Push/lib/Connector/AMQP.pm | 230 |
1 files changed, 0 insertions, 230 deletions
diff --git a/extensions/Push/lib/Connector/AMQP.pm b/extensions/Push/lib/Connector/AMQP.pm deleted file mode 100644 index 7b7d4aa72..000000000 --- a/extensions/Push/lib/Connector/AMQP.pm +++ /dev/null @@ -1,230 +0,0 @@ -# This Source Code Form is subject to the terms of the Mozilla Public -# License, v. 2.0. If a copy of the MPL was not distributed with this -# file, You can obtain one at http://mozilla.org/MPL/2.0/. -# -# This Source Code Form is "Incompatible With Secondary Licenses", as -# defined by the Mozilla Public License, v. 2.0. - -package Bugzilla::Extension::Push::Connector::AMQP; - -use strict; -use warnings; - -use base 'Bugzilla::Extension::Push::Connector::Base'; - -use Bugzilla::Constants; -use Bugzilla::Extension::Push::Constants; -use Bugzilla::Extension::Push::Util; -use Bugzilla::Util qw(generate_random_password); -use DateTime; - -sub init { - my ($self) = @_; - $self->{mq} = 0; - $self->{channel} = 1; - - if ($self->config->{queue}) { - $self->{queue_name} = $self->config->{queue}; - } else { - my $queue_name = Bugzilla->params->{'urlbase'}; - $queue_name =~ s#^https?://##; - $queue_name =~ s#/$#|#; - $queue_name .= generate_random_password(16); - $self->{queue_name} = $queue_name; - } -} - -sub options { - return ( - { - name => 'host', - label => 'AMQP Hostname', - type => 'string', - default => 'localhost', - required => 1, - }, - { - name => 'port', - label => 'AMQP Port', - type => 'string', - default => '5672', - required => 1, - validate => sub { - $_[0] =~ /\D/ && die "Invalid port (must be numeric)\n"; - }, - }, - { - name => 'username', - label => 'Username', - type => 'string', - default => 'guest', - required => 1, - }, - { - name => 'password', - label => 'Password', - type => 'password', - default => 'guest', - required => 1, - }, - { - name => 'vhost', - label => 'Virtual Host', - type => 'string', - default => '/', - required => 1, - }, - { - name => 'exchange', - label => 'Exchange', - type => 'string', - default => '', - required => 1, - }, - { - name => 'queue', - label => 'Queue', - type => 'string', - }, - ); -} - -sub stop { - my ($self) = @_; - if ($self->{mq}) { - Bugzilla->push_ext->logger->debug('AMQP: disconnecting'); - $self->{mq}->disconnect(); - $self->{mq} = 0; - } -} - -sub _connect { - my ($self) = @_; - my $logger = Bugzilla->push_ext->logger; - my $config = $self->config; - - $self->stop(); - - $logger->debug('AMQP: Connecting to RabbitMQ ' . $config->{host} . ':' . $config->{port}); - require Net::RabbitMQ; - my $mq = Net::RabbitMQ->new(); - $mq->connect( - $config->{host}, - { - port => $config->{port}, - user => $config->{username}, - password => $config->{password}, - } - ); - $self->{mq} = $mq; - - $logger->debug('AMQP: Opening channel ' . $self->{channel}); - $self->{mq}->channel_open($self->{channel}); - - $logger->debug('AMQP: Declaring queue ' . $self->{queue_name}); - $self->{mq}->queue_declare( - $self->{channel}, - $self->{queue_name}, - { - passive => 0, - durable => 1, - exclusive => 0, - auto_delete => 0, - }, - ); -} - -sub _bind { - my ($self, $message) = @_; - my $logger = Bugzilla->push_ext->logger; - my $config = $self->config; - - # bind to queue (also acts to verify the connection is still valid) - if ($self->{mq}) { - eval { - $logger->debug('AMQP: binding queue(' . $self->{queue_name} . ') with exchange(' . $config->{exchange} . ')'); - $self->{mq}->queue_bind( - $self->{channel}, - $self->{queue_name}, - $config->{exchange}, - $message->routing_key, - ); - }; - if ($@) { - $logger->debug('AMQP: ' . clean_error($@)); - $self->{mq} = 0; - } - } - -} - -sub should_send { - my ($self, $message) = @_; - my $logger = Bugzilla->push_ext->logger; - - my $payload = $message->payload_decoded(); - my $target = $payload->{event}->{target}; - my $is_private = $payload->{$target}->{is_private} ? 1 : 0; - if (!$is_private && exists $payload->{$target}->{bug}) { - $is_private = $payload->{$target}->{bug}->{is_private} ? 1 : 0; - } - - if ($is_private) { - # we only want to push the is_private message from the change_set, as - # this is guaranteed to contain public information only - if ($message->routing_key !~ /\.modify:is_private$/) { - $logger->debug('AMQP: Ignoring private message'); - return 0; - } - $logger->debug('AMQP: Sending change of message to is_private'); - } - return 1; -} - -sub send { - my ($self, $message) = @_; - my $logger = Bugzilla->push_ext->logger; - my $config = $self->config; - - # don't push comments to pulse - if ($message->routing_key =~ /^comment\./) { - $logger->debug('AMQP: Ignoring comment'); - return PUSH_RESULT_IGNORED; - } - - # don't push private data - $self->should_push($message) - || return PUSH_RESULT_IGNORED; - - $self->_bind($message); - - eval { - # reconnect if required - if (!$self->{mq}) { - $self->_connect(); - } - - # send message - $logger->debug('AMQP: Publishing message'); - $self->{mq}->publish( - $self->{channel}, - $message->routing_key, - $message->payload, - { - exchange => $config->{exchange}, - }, - { - content_type => 'text/plain', - content_encoding => '8bit', - }, - ); - }; - if ($@) { - return (PUSH_RESULT_TRANSIENT, clean_error($@)); - } - - return PUSH_RESULT_OK; -} - -1; - |