File Coverage

blib/lib/POE/Filter/Zlib/Stream.pm

Criterion	Covered	Total	%
statement	71	83	85.5
branch	17	26	65.3
condition	2	3	66.6
subroutine	11	12	91.6
pod	6	6	100.0
total	107	130	82.3

package POE::Filter::Zlib::Stream;

use strict;
use warnings;
use Carp;
use Compress::Raw::Zlib qw(Z_OK Z_STREAM_END Z_FINISH Z_SYNC_FLUSH);
use vars qw($VERSION);
use base qw(POE::Filter);

$VERSION = '2.02';

sub new {
  my $type = shift;
  croak "$type requires an even number of parameters" if @_ % 2;
  my $buffer = { @_ };
  $buffer->{ lc $_ } = delete $buffer->{ $_ } for keys %{ $buffer };
  $buffer->{BUFFER} = '';
  delete $buffer->{deflateopts} unless ref ( $buffer->{deflateopts} ) eq 'HASH';
  $buffer->{d} = Compress::Raw::Zlib::Deflate->new( %{ $buffer->{deflateopts} } );
  unless ( $buffer->{d} ) {
	warn "Failed to create deflate stream\n";
	return;
  }
  delete $buffer->{inflateopts} unless ref ( $buffer->{inflateopts} ) eq 'HASH';
  $buffer->{i} = Compress::Raw::Zlib::Inflate->new ( %{ $buffer->{inflateopts} } );
  unless ( $buffer->{i} ) {
	warn "Failed to create inflate stream\n";
	return;
  }
  if (not defined $buffer->{flushtype}) {
  	$buffer->{flushtype} = Z_SYNC_FLUSH;
  }
  return bless $buffer, $type;
}

# use inherited get() from POE::Filter

sub get_one_start {
  my ($self, $raw_lines) = @_;
  $self->{BUFFER} .= join '', @{ $raw_lines };
}

sub get_one {
  my $self = shift;

  return [ ] unless length $self->{BUFFER};
  my ($status, $out);
  $status = $self->{i}->inflate( \$self->{BUFFER}, $out );

  unless ( $status == Z_OK or $status == Z_STREAM_END ) {
    warn "Couldn\'t inflate buffer\n";
    return [ ];
  }
  if ($status == Z_STREAM_END) {
    $self->{i} = Compress::Raw::Zlib::Inflate->new ( %{ $self->{inflateopts} } );
  }
  return [ $out ];
}

sub get_pending {
  my $self = shift;
  return $self->{BUFFER} ? [ $self->{BUFFER} ] : undef;
}

sub put {
  my ($self, $events) = @_;
  my $raw_lines = [];

  foreach my $event (@$events) {
	my ($dstat, $dout);
	$dstat = $self->{d}->deflate( $event, $dout );
	unless ( $dstat == Z_OK ) {
	  warn "(data) Couldn\'t deflate: $event\n($dstat)";
	  next;
	}
	my ($fout,$fstat);
	$fstat = $self->{d}->flush( $fout, $self->{flushtype} );
	unless ( $fstat == Z_OK ) {
	  warn "(flush) Couldn\'t flush/deflate: $event\n";
	  next;
	}
	if ($self->{flushtype} == Z_FINISH) {
  	  $self->{d} = Compress::Raw::Zlib::Deflate->new ( %{ $self->{deflateopts} } );
	}
	push @$raw_lines, $dout . $fout;
  }
  return $raw_lines;
}

sub clone {
  my $self = shift;
  my $nself = { };
  $nself->{$_} = $self->{$_} for keys %{ $self };
  $nself->{d} = Compress::Raw::Zlib::Deflate->new( %{ $nself->{deflateopts} } );
  $nself->{i} = Compress::Raw::Zlib::Inflate->new( %{ $nself->{inflateopts} } );
  $nself->{BUFFER} = '';
  return bless $nself, ref $self;
}

1;

__END__