view release on metacpan or  search on metacpan

lib/App/Chart/Glib/Ex/DirBroadcast.pm  view on Meta::CPAN

      $subr->($obj, @_);
    } else {
      _disconnect ($self, $key, $csubr);
    }
  };
  $self->connect ($key, $csubr);
}

sub _disconnect {
  my ($self, $key, $subr) = @_;
  if (my $aref = $self->{'connections'}->{$key}) {
    @$aref = grep {$_ != $subr} @$aref;
  }
}

sub send_locally {
  my ($self, $key, @data) = @_;
  ref($self) or $self = $self->instance;

  if ($self->{'hold'}) {
    push @{$self->{'hold_list'}}, sub { send_locally ($self, $key, @data); };

  } else {
    if (my $aref = $self->{'connections'}->{$key}) {
      foreach my $subr (@$aref) {
        $subr->(@data);
      }
    }
  }
}

sub listen {
  my ($self) = @_;
  ref($self) or $self = $self->instance;

  if ($self->{'listen_source_ids'}) { return; }  # already done
  my $directory = $self->{'directory'};
  if (! defined $directory) {
    croak 'DirBroadcast cannot listen until broadcast directory is set';
  }

  require File::Path;
  File::Path::mkpath ($directory);

  require Sys::Hostname;
  my $hostname = Sys::Hostname::hostname();
  my $listen_filename = $self->{'listen_filename'}
    = File::Spec->catfile ($directory, "$hostname.$$");
  unlink ($listen_filename);  # possible previous leftover

  # as usual socket() and friends get FD_CLOEXEC set automatically, no need
  # to do anything special to avoid propagating $listen_sock fd down to
  # subprocess jobs
  require Socket;
  require IO::Socket;
  my $listen_sock = $self->{'listen_sock'}
    = do { local $^F = 0; # ensure close-on-exec for the socket
           IO::Socket->new (Domain => Socket::AF_UNIX(),
                            Type   => Socket::SOCK_DGRAM(),
                            Local  => $listen_filename) };
  binmode ($listen_sock, ':raw') or die;
  ### DirBroadcast listen: $listen_filename, $listen_sock->fileno

  require Glib;
  require App::Chart::Glib::Ex::MoreUtils;
  require Glib::Ex::SourceIds;

  $self->{'listen_source_ids'}
    = Glib::Ex::SourceIds->new
      (Glib::IO->add_watch ($listen_sock->fileno,
                            ['in', 'hup', 'err'],
                            \&_do_listen_read,
                            App::Chart::Glib::Ex::MoreUtils::ref_weak($self)));
}

sub _do_listen_read {
  my ($fd, $conditions, $ref_weak_self) = @_;
  my $self = $$ref_weak_self || return 0; # Glib::SOURCE_REMOVE

  ### DirBroadcast read: $fd, $conditions
  my $buf;
  my $ret = $self->{'listen_sock'}->recv ($buf, MAXLEN, 0);
  if (! defined $ret) {
    warn __PACKAGE__." listen read error: $!";
    delete $self->{'listen_source_ids'};
    delete $self->{'listen_sock'};
    return 0; # Glib::SOURCE_REMOVE
  }
  ### receive: " bytes=".length($buf), "'$ret'"
  require Storable;
  my $args;
  if (! eval { $args = Storable::thaw ($buf); 1 }) {
    carp "DirBroadcast: error thawing message, ignoring";
    return 1; # Glib::SOURCE_CONTINUE
  }

  ### $args
  $self->send_locally (@$args);
  return 1; # Glib::SOURCE_CONTINUE
}

my $send_sock;

sub send {
  my ($self, $key, @data) = @_;
  ref($self) or $self = $self->instance;

  if ($self->{'hold'}) {
    ### send hold back: $key
    push @{$self->{'hold_list'}}, sub { $self->send ($key, @data); };
    return;
  }
  #### send: $key, \@data

  $self->send_locally ($key, @data);

  my $directory = $self->{'directory'};
  require IO::Dir;
  my $dh = IO::Dir->new ($directory) || return;
  my @filenames = $dh->read;
  $dh->close;

  my $pattern = ($self->{'pattern'} ||= do {
    require Sys::Hostname;
    my $hostname = Sys::Hostname::hostname();
    qr/^\Q$hostname\E\.[0-9]+$/o
  });

  my $frozen;
  foreach my $filename (@filenames) {
    if ($filename !~ $pattern) { next; }
    $filename = File::Spec->catfile ($directory, $filename);

    if ($filename eq ($self->{'listen_filename'}||'')) {
      next;  # ourselves
    }
    ### DirBroadcast to: $filename

    # send_sock created on first of any DirBroadcast instance and then kept
    # open
    $send_sock ||= do {
      require IO::Socket;
      require Socket;
      my $sock = do {
        local $^F = 0; # ensure close-on-exec for the socket
        IO::Socket->new (Domain => Socket::AF_UNIX(),
                         Type   => Socket::SOCK_DGRAM());
      };
      $sock->blocking(0);
      binmode ($sock, ':raw') or die;
      $sock
    };

    # put off freezing until we find someone to send to
    if (! defined $frozen) {
      require Storable;
      $frozen = Storable::freeze ([$key, @data]);
      if (length ($frozen) > MAXLEN) {
        croak 'DirBroadcast: message too long: ',length($frozen);
      }
    }

    my $sun = Socket::sockaddr_un ($filename);
    my $sent = $send_sock->send ($frozen, 0, $sun);
    if (! defined $sent || $sent != length($frozen)) {
      ### send: (! defined $sent && "removing, error $!") || "removing, short send $sent bytes"
      unlink ($filename);
    }
  }
}

sub hold {
  my ($self) = @_;
  ref($self) or $self = $self->instance;
  return App::Chart::Glib::Ex::DirBroadcast::Hold->new ($self);
}

package App::Chart::Glib::Ex::DirBroadcast::Hold;
use strict;
use warnings;

sub new {
  my ($class, $dirb) = @_;
  my $self = bless { }, $class;
  $self->{'target'} = $dirb;
  require Scalar::Util;
  Scalar::Util::weaken ($self->{'target'});
  $dirb->{'hold'} ++;
  return $self;
}

sub DESTROY {
  my ($self) = @_;
  my $dirb = delete $self->{'target'} || return;
  if (-- $dirb->{'hold'}) { return; }

  my $hold_list = $dirb->{'hold_list'};
  ### DirBroadcast::Hold now run: $hold_list
  while (my $subr = shift @$hold_list) {
    $subr->();
  }
}

1;
__END__

=head1 NAME

App::Chart::Glib::Ex::DirBroadcast -- broadcast messages through a directory of named pipes