xmit_kstat

#!/usr/bin/env perl

use v5.18.1;
use strict;
use warnings;

use Net::Async::AMQP;
use IO::Async::Loop;
use Data::Dumper;
use Future::Utils qw( repeat try_repeat );

# Detect disconnection from AMQP
my $DISCONNECTED;
# The channel we're publishing on
my $CHANNEL;
# The exchange we're publishing on
my $EXCHANGE;

my $loop = IO::Async::Loop->new;

$loop->add(
  my $mq = Net::Async::AMQP->new()
);

my $true = Net::AMQP::Value->true;
my $msg  = 'info MESSAGE';

$mq->connect(
  #host      => '192.168.56.1',
  host      => 'localhost',
  user      => 'solperf',
  pass      => 'solperf',
  vhost     => '/',
  client_properties => {
    capabilities => {
      'consumer_cancel_notify' => $true,
      'connection.blocked'     => $true,
    },
  },
)->then(
  sub {
    shift->open_channel;
  }, sub { say "CONNECT FAILED!" }
)->on_done(sub { say "CONNECTED!"; })->then(
  sub {
    #my ($channel)  = shift;
    $CHANNEL = shift;
    #say "GOT CHANNEL: " . Dump($CHANNEL);
    say "GOT CHANNEL: $CHANNEL";

    $EXCHANGE = 'logs';

    Future->needs_all(
      # This is subscriber code
      $CHANNEL->queue_declare(
        queue     => '',  # Let the queue name be generated
        exclusive => 1,
      ),
      # This is publisher code
      $CHANNEL->exchange_declare(
        type     => 'fanout',
        exchange => $EXCHANGE,
      )
    )->on_done(sub { say "EXCHANGE DECLARED!"; })->then(
      sub {
        my ($q) = shift;
        say "QUEUE IS: $q";

        # Register for events of interest
        $CHANNEL->bus->subscribe_to_event(
          connected => sub {
            say "CONNECTED TO AMQP";
          },
          close => sub {
            say "DETECTED AMQP CLOSING OF CONNECTION";
            $DISCONNECTED = 1;
          },
          heartbeat_failure => sub {
            say "HEARTBEAT FAILURE DETECTED";
          },
          unexpected_frame => sub {
            say "UNEXPECTED FRAME";
          },
        );

        $q->bind_exchange(
          channel  => $CHANNEL,
          exchange => $EXCHANGE,
          routing_key => '',
        )->on_done( sub { say "EXCHANGE BOUND"; } );
      },
      sub { say "EXCHANGE DECLARE FAILED"; }
    );
  },
  sub { say "OPEN CHANNEL FAILED"; }
)->then(
  sub {
    my ($c) = 0;
    try_repeat {
      say "PUBLISHING $msg $c";
      $c++;
      $CHANNEL->publish(
        exchange    => $EXCHANGE,
        routing_key => '',
        #type        => 'application/json',
        type        => 'text/plain',
        payload     => $msg,
      )->get;
      sleep 1;
    } until => sub { $DISCONNECTED };
  },
  sub { say "Problem with Exchange Declare or Publish"; }
)->get;


$loop->run;