view release on metacpan or  search on metacpan

lib/Beekeeper/Client.pm  view on Meta::CPAN

                unless (defined $method && $method =~ m/^([\.\w-]+)\.([\w-]+)$/) {
                    warn "Received notification with invalid method '$method' $at";
                    return;
                }

                my $cb = $client->{callbacks}->{"msg.$1.$2"} || 
                         $client->{callbacks}->{"msg.$1.*"};

                unless ($cb) {
                    warn "No callback found for received notification '$method' $at";
                    return;
                }

                $cb->($resp->{params}, $resp);
            }
        },
        on_suback => sub {
            my ($success, $prop) = @_;
            die "Could not subscribe to response topic '$response_topic' $at" unless $success;
        }
    );

    return $response_topic;
}

sub wait_async_calls {
    my ($self) = @_;

    # Wait for all pending async requests
    my $cv = delete $self->{_CLIENT}->{async_cv};
    return unless defined $cv;

    # Make AnyEvent to allow one level of recursive condvar blocking, as we may
    # block both in $worker->__work_forever and here
    $AE_WAITING && Carp::confess "Recursive condvar blocking wait attempted";
    local $AE_WAITING = 1;
    local $AnyEvent::CondVar::Base::WAITING = 0;

    $cv->recv;
}


sub get_authentication_data {
    my ($self) = @_;

    $self->{_CLIENT}->{auth_data};
}

sub set_authentication_data {
    my ($self, $data) = @_;

    $self->{_CLIENT}->{auth_data} = $data;
}

sub __use_authorization_token {
    my ($self, $token) = @_;

    # Using a hashing function makes harder to access the wrong worker pool by mistake,
    # but it is not an effective access restriction: anyone with access to the backend
    # bus credentials can easily inspect and clone auth data tokens

    my $salt = $self->{_CLIENT}->{auth_salt};

    my $adata_ref = \$self->{_CLIENT}->{auth_data};

    my $guard = Beekeeper::Client::Guard->new( $adata_ref );

    $$adata_ref = md5_base64($token . $salt);

    return $guard;
}

1;

package
    Beekeeper::Client::Guard;   # hide from PAUSE

sub new {
    my ($class, $ref) = @_;

    bless [$ref, $$ref], $class;
}

sub DESTROY {

    ${$_[0]->[0]} = $_[0]->[1];
}

1;

__END__

=pod

=encoding utf8

=head1 NAME
 
Beekeeper::Client - Make RPC calls through message bus

=head1 VERSION
 
Version 0.09

=head1 SYNOPSIS

  my $client = Beekeeper::Client->instance;
  
  $client->send_notification(
      method => "my.service.foo",
      params => { foo => $foo },
  );
  
  my $resp = $client->call_remote(
      method => "my.service.bar",
      params => { %args },
  );
  
  die uneless $resp->success;