Rename AfterFirst to AfterAny for consistency with Java.
Project: http://git-wip-us.apache.org/repos/asf/beam/repo Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/2c347192 Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/2c347192 Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/2c347192 Branch: refs/heads/master Commit: 2c347192709e4076a1ac59b5ea769d4d3b2494f4 Parents: 4b9029a Author: Robert Bradshaw <[email protected]> Authored: Tue Apr 11 11:00:04 2017 -0700 Committer: Robert Bradshaw <[email protected]> Committed: Thu Apr 20 08:55:05 2017 -0700 ---------------------------------------------------------------------- sdks/python/apache_beam/transforms/trigger.py | 10 ++++++---- sdks/python/apache_beam/transforms/trigger_test.py | 10 +++++----- 2 files changed, 11 insertions(+), 9 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/beam/blob/2c347192/sdks/python/apache_beam/transforms/trigger.py ---------------------------------------------------------------------- diff --git a/sdks/python/apache_beam/transforms/trigger.py b/sdks/python/apache_beam/transforms/trigger.py index 6a4cf24..b9786f4 100644 --- a/sdks/python/apache_beam/transforms/trigger.py +++ b/sdks/python/apache_beam/transforms/trigger.py @@ -186,7 +186,7 @@ class TriggerFn(object): def from_runner_api(proto, context): return { 'after_all': AfterAll, - 'after_any': AfterFirst, + 'after_any': AfterAny, 'after_each': AfterEach, 'after_end_of_window': AfterWatermark, # after_processing_time, after_synchronized_processing_time @@ -488,7 +488,8 @@ class ParallelTriggerFn(TriggerFn): in proto.after_all.subtriggers or proto.after_any.subtriggers] if proto.after_all.subtriggers: return AfterAll(*subtriggers) - return AfterFirst(*subtriggers) + else: + return AfterAny(*subtriggers) def to_runner_api(self, context): subtriggers = [ @@ -505,7 +506,7 @@ class ParallelTriggerFn(TriggerFn): raise NotImplementedError(self) -class AfterFirst(ParallelTriggerFn): +class AfterAny(ParallelTriggerFn): """Fires when any subtrigger fires. Also finishes when any subtrigger finishes. @@ -589,7 +590,8 @@ class AfterEach(TriggerFn): for subtrigger in self.triggers])) -class OrFinally(AfterFirst): +class OrFinally(AfterAny): + @staticmethod def from_runner_api(proto, context): return OrFinally( http://git-wip-us.apache.org/repos/asf/beam/blob/2c347192/sdks/python/apache_beam/transforms/trigger_test.py ---------------------------------------------------------------------- diff --git a/sdks/python/apache_beam/transforms/trigger_test.py b/sdks/python/apache_beam/transforms/trigger_test.py index 9f2046a..914babb 100644 --- a/sdks/python/apache_beam/transforms/trigger_test.py +++ b/sdks/python/apache_beam/transforms/trigger_test.py @@ -33,7 +33,7 @@ from apache_beam.transforms.trigger import AccumulationMode from apache_beam.transforms.trigger import AfterAll from apache_beam.transforms.trigger import AfterCount from apache_beam.transforms.trigger import AfterEach -from apache_beam.transforms.trigger import AfterFirst +from apache_beam.transforms.trigger import AfterAny from apache_beam.transforms.trigger import AfterWatermark from apache_beam.transforms.trigger import DefaultTrigger from apache_beam.transforms.trigger import GeneralTriggerDriver @@ -217,7 +217,7 @@ class TriggerTest(unittest.TestCase): def test_fixed_after_first(self): self.run_trigger_simple( FixedWindows(10), # pyformat break - AfterFirst(AfterCount(2), AfterWatermark()), + AfterAny(AfterCount(2), AfterWatermark()), AccumulationMode.ACCUMULATING, [(1, 'a'), (2, 'b'), (3, 'c')], {IntervalWindow(0, 10): [set('ab')]}, @@ -225,7 +225,7 @@ class TriggerTest(unittest.TestCase): 2) self.run_trigger_simple( FixedWindows(10), # pyformat break - AfterFirst(AfterCount(5), AfterWatermark()), + AfterAny(AfterCount(5), AfterWatermark()), AccumulationMode.ACCUMULATING, [(1, 'a'), (2, 'b'), (3, 'c')], {IntervalWindow(0, 10): [set('abc')]}, @@ -236,7 +236,7 @@ class TriggerTest(unittest.TestCase): def test_repeatedly_after_first(self): self.run_trigger_simple( FixedWindows(100), # pyformat break - Repeatedly(AfterFirst(AfterCount(3), AfterWatermark())), + Repeatedly(AfterAny(AfterCount(3), AfterWatermark())), AccumulationMode.ACCUMULATING, zip(range(7), 'abcdefg'), {IntervalWindow(0, 100): [ @@ -388,7 +388,7 @@ class RunnerApiTest(unittest.TestCase): for trigger_fn in ( DefaultTrigger(), AfterAll(AfterCount(1), AfterCount(10)), - AfterFirst(AfterCount(10), AfterCount(100)), + AfterAny(AfterCount(10), AfterCount(100)), AfterWatermark(early=AfterCount(1000)), AfterWatermark(early=AfterCount(1000), late=AfterCount(1)), Repeatedly(AfterCount(100)),
