File Coverage

blib/lib/Mojo/RabbitMQ/Client/Method/Publish.pm

Criterion	Covered	Total	%
statement	3	29	10.3
branch	0	4	0.0
condition	0	11	0.0
subroutine	1	7	14.2
pod	2	2	100.0
total	6	53	11.3

package Mojo::RabbitMQ::Client::Method::Publish;
use Mojo::Base 'Mojo::RabbitMQ::Client::Method';

sub setup {
  my $self = shift;

  $self->arguments({@_});

  return $self;
}

sub deliver {
  my $self = shift;

  return $self if !$self->channel->is_active;

  my %args = %{$self->arguments};

  my $header_args
    = {header => delete $args{header} || {}, weight => delete $args{weight}};
  my $body = delete $args{body} || '';

  $self->_publish(%args)->_header($header_args, $body)->_body($body, sub {
    $self->is_sent(1);

    $self->emit('success');
  });

  return $self if !$args{mandatory} && !$args{immediate};

  $self->channel->return_cbs->{($args{exchange} || '') . '_'
      . $args{routing_key}} = $self;

  return $self;
}

sub _publish {
  my $self = shift;
  my %args = @_;

  $self->client->_write_frame(
    Net::AMQP::Protocol::Basic::Publish->new(
      exchange  => '',
      mandatory => 0,
      immediate => 0,
      %args,    # routing_key
      ticket => 0,
    ),
    $self->channel->id
  );

  return $self;
}

sub _header {
  my ($self, $args, $body) = @_;

  $self->client->_write_frame(
    Net::AMQP::Frame::Header->new(
      weight => $args->{weight} || 0,
      body_size    => length($body),
      header_frame => Net::AMQP::Protocol::Basic::ContentHeader->new(
        content_type     => 'application/octet-stream',
        content_encoding => undef,
        headers          => {},
        delivery_mode    => 1,
        priority         => 1,
        correlation_id   => undef,
        expiration       => undef,
        message_id       => undef,
        timestamp        => time,
        type             => undef,
        user_id          => $self->client->user,
        app_id           => undef,
        cluster_id       => undef,
        %{ $args->{header} },
      ),
    ),
    $self->channel->id
  );

  return $self;
}

sub _body {
  my ($self, $body, $cb) = @_;

  $self->client->_write_frame(Net::AMQP::Frame::Body->new(payload => $body),
    $self->channel->id, $cb);

  return $self;
}

1;

=encoding utf8

=head1 NAME

Mojo::RabbitMQ::Client::Method::Publish - single class to do all of AMQP Publish method magic

=head1 SYNOPSIS

  use Mojo::RabbitMQ::Client::Method::Publish;

  my $method = Mojo::RabbitMQ::Client::Method::Publish->new(
    client => $client,
    channel => $channel
  )->setup(
    exchange    => 'mojo',
    routing_key => '',
    header      => {}
    body        => 'mojo',
    mandatory   => 0,
    immediate   => 0,
  )->deliver();

=head1 DESCRIPTION

L is a special class to implement AMQP message publish workflow.

=head1 EVENTS

L inherits all events from L.

=head1 ATTRIBUTES

L inherits all attributes from L.

=head1 METHODS

L inherits all methods from L with
following changes.

=head2 setup

  $method = $method->setup($arguments);

Only accepts common arguments for message publish chain. Which is:

=over 2

=item Frame::Method

=over 2

=item Basic::Publish

=over 2

=item * exchange

=item * routing_key

=item * mandatory

=item * immediate

=back

=back

=item Frame::Header

=over 2

=item Basic::ContentHeader

=over 2

=item * header

=item * weight

=back

=back

=item Frame::Body

=over 2

=item * body (as payload)

=back

=back

=head1 SEE ALSO

L, L, L

=head1 COPYRIGHT AND LICENSE

Copyright (C) 2015-2017, Sebastian Podjasek and others

Based on L - Copyright (C) 2010 Masahito Ikuta, maintained by C<< [e-mail] >>

This program is free software, you can redistribute it and/or modify it under the terms of the Artistic License version 2.0.

=cut