view release on metacpan or  search on metacpan

t/slide.t  view on Meta::CPAN

ok ($@, "expect 'too many columns' error");

# wrapping cases
is_deeply([9,1], [$c->destraddle( 1,10)],  "wrap: 1 + 10 -> 9, 1");
is_deeply([8,1], [$c->destraddle( 2, 9)],  "wrap: 2 + 9  -> 8, 1");
is_deeply([1,1], [$c->destraddle( 9, 2)],  "wrap: 9 + 2  -> 1, 1");
is_deeply([9,1], [$c->destraddle(11,10)],  "wrap: 11 + 10 -> 9, 1");
is_deeply([8,1], [$c->destraddle(12, 9)],  "wrap: 12 + 9  -> 8, 1");
is_deeply([1,1], [$c->destraddle(19, 2)],  "wrap: 19 + 2  -> 1, 1");

# test advance of single read stream (split)

ok($s->advance_read(1), "advance_read(1) returns true");
($rok,$pok,$wok,$subs) = $s->can_advance;
is ($rok, 9, "read fills single input buffer");
is ($pok, 1, "read allows a column to be processed");
is ($wok, 0, "read is independent of write");

# fill up the single input buffer
$s->advance_read(9);
($rok,$pok,$wok,$subs) = $s->can_advance;
is ($rok, 0,  "single input buffer full");
is ($pok, 10, "read allowed 10 columns to be processed");
is ($wok, 0,  "read is independent of write");

eval {$s->advance_read(1)}; ok($@, "expect error reading past window");

# test advance of bundled read streams (combine)

# Try to get coverage for all possible scenarios...
#
# 1. advance row 0 by 1
# 2. advance row 1 by 4
# 3. advance row 2 by 6 (triggering update of parent; leapfrog)
#
# 4. advance row 0 by 1 (becoming 2, updating parent; no leapfrog)
# 5. advance row 0 by 4 (becoming 6, updating parent, now with 2 rows @ 6)
# 6. advance row 1 by 2 (all at 6 now)
#
# I'll test both can_advance outputs and introspect into parent
# pointer and yts values at the start, but drop some of that testing
# once it's clear that they're being updated in sync.

# we should be able to add a callback even after object construction
# thanks to Class::Tiny creating accessors
my $read_cb_count = 0;
ok($c->cb_read_bundle(sub { ++$read_cb_count}),
   "Added callback for when parent's read pointer updates");

# 1. advance row 0 by 1
ok (0 == $c->advance_read_substream(0,1), "can advance row 0 by 1");
($rok,$pok,$wok,$subs) = $c->can_advance;

is ($rok, 10, "expect read one substream not to advance parent");
is ($pok, 0,  "expect read one substream not to allow processing");
is ($wok, 0,  "expect read independent of write");
is ($subs->[0], 9, "expect reduced read space in substream");
is ($c->bundle->[0]->{head}, 1, "advanced substream's head variable");
is ($c->yts, 2, "expect read substream to decrement 'yet to start'");

is ($read_cb_count, 0, "expect no read callbacks yet");

eval { $c->advance_read_substream(0,10) };
ok ($@, "expect error advancing substream past window");

# 2. advance row 1 by 4
ok (0 == $c->advance_read_substream(1,4), "can advance row 1 by 4");
($rok,$pok,$wok,$subs) = $c->can_advance;

is ($rok, 10, "expect read two substreams not to advance parent");
is ($pok, 0,  "expect read two substream not to allow processing");
is ($wok, 0,  "expect read still independent of write");
is ($subs->[1], 6, "expect reduced read space in substream");
is ($c->bundle->[1]->{head}, 4, "advanced substream's head variable");
is ($c->yts, 1, "expect read substream to decrement 'yet to start'");

is ($read_cb_count, 0, "expect no read callbacks yet");

# 3. advance row 2 by 6 (triggering update of parent)
ok(1 == $c->advance_read_substream(2,6), "can advance row 2 by 6");
($rok,$pok,$wok,$subs) = $c->can_advance;

is ($rok, 9,  "expect read 3 substreams advance parent by min");
is ($pok, 1,  "expect read 3 substreams to allow processing");
is ($wok, 0,  "expect read still independent of write");
is ($subs->[2], 4, "expect reduced read space in substream");
is ($c->bundle->[2]->{head}, 6, "advanced substream's head variable");
is ($c->bundle->[2]->{tail}, 0, "expect substream's tail unchanged");
is ($c->yts, 1, "expect new 'yet to start': 1 stream at min");

is ($read_cb_count, 1, "expected 1st read callback");

# 4. advance row 0 by 1 (becoming 2, updating parent; no leapfrog)
ok(1 == $c->advance_read_substream(0,1), "can advance row 0 by 1 to 2");
($rok,$pok,$wok,$subs) = $c->can_advance;

is ($rok, 8,  "expect read space reduced by 1 again");
is ($pok, 2,  "expect another column of processing");
is ($subs->[0], 8, "expect reduced read space in substream");
is ($c->bundle->[0]->{head}, 2, "advanced substream's head variable");
is ($c->yts, 1, "expect no other streams at read level 2");

is ($read_cb_count, 2, "expected 2nd read callback");

# 5. advance row 0 by 4 (becoming 6, updating parent, now with 2 rows @ 6)
# (still waiting for stream 1 @ 4 to catch up)
ok(1 == $c->advance_read_substream(0,4), "can advance row 0 by 4 to 6");
($rok,$pok,$wok,$subs) = $c->can_advance;

is ($rok, 6,  "expect min read fill is row 1, with 4 read");
is ($pok, 4,  "expect another 2 columns of processing");
is ($subs->[0], 4, "expect reduced read space in substream");
is ($c->bundle->[0]->{head}, 6, "advanced substream's head variable");
is ($c->yts, 1, "expect waiting on laggard stream 1");

is ($read_cb_count, 3, "expected 3rd read callback");

# 6. advance row 1 by 2 (all at 6 now)
ok(1 == $c->advance_read_substream(1,2), "can advance row 1 by 2 to 6");
($rok,$pok,$wok,$subs) = $c->can_advance;

is ($rok, 4,  "expect all read substreams at 6 reads");
is ($pok, 6,  "expect another 2 columns to process");
is ($c->yts, 3, "expect all streams waiting together");

is ($read_cb_count, 4, "expected 4th read callback");

# Testing write substreams (split) needn't be so comprehensive, but we
# will need to check a few things:
#
# * it's updating the tail/write_tail pointer
# * interactions with processing (also similar for read buffer)
# * that we can advance read_head two windows from write_tail
# * that it's using the correct callback
#
# First, though, testing advance_process using our current combiner
# since it has 6 data columns ready to go.