You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by ke...@apache.org on 2017/02/06 16:53:44 UTC
[2/4] beam git commit: Autoformat
GroupAlsoByWindowsViaOutputBufferDoFn
Autoformat GroupAlsoByWindowsViaOutputBufferDoFn
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/c2925feb
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/c2925feb
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/c2925feb
Branch: refs/heads/master
Commit: c2925feb11ce26308e82209929ea478a93e06c84
Parents: 4800a3e
Author: Kenneth Knowles <kl...@google.com>
Authored: Mon Feb 6 08:30:31 2017 -0800
Committer: Kenneth Knowles <kl...@google.com>
Committed: Mon Feb 6 08:30:31 2017 -0800
----------------------------------------------------------------------
.../GroupAlsoByWindowsViaOutputBufferDoFn.java | 37 ++++++++++----------
1 file changed, 19 insertions(+), 18 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/beam/blob/c2925feb/runners/core-java/src/main/java/org/apache/beam/runners/core/GroupAlsoByWindowsViaOutputBufferDoFn.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/GroupAlsoByWindowsViaOutputBufferDoFn.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/GroupAlsoByWindowsViaOutputBufferDoFn.java
index d83060f..bc02679 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/GroupAlsoByWindowsViaOutputBufferDoFn.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/GroupAlsoByWindowsViaOutputBufferDoFn.java
@@ -35,7 +35,7 @@ import org.joda.time.Instant;
*/
@SystemDoFnInternal
public class GroupAlsoByWindowsViaOutputBufferDoFn<K, InputT, OutputT, W extends BoundedWindow>
- extends GroupAlsoByWindowsDoFn<K, InputT, OutputT, W> {
+ extends GroupAlsoByWindowsDoFn<K, InputT, OutputT, W> {
private final WindowingStrategy<?, W> strategy;
private final StateInternalsFactory<K> stateInternalsFactory;
@@ -90,25 +90,26 @@ public class GroupAlsoByWindowsViaOutputBufferDoFn<K, InputT, OutputT, W extends
reduceFnRunner.persist();
}
- private void fireEligibleTimers(InMemoryTimerInternals timerInternals,
- ReduceFnRunner<K, InputT, OutputT, W> reduceFnRunner) throws Exception {
+ private void fireEligibleTimers(
+ InMemoryTimerInternals timerInternals, ReduceFnRunner<K, InputT, OutputT, W> reduceFnRunner)
+ throws Exception {
List<TimerInternals.TimerData> timers = new ArrayList<>();
while (true) {
- TimerInternals.TimerData timer;
- while ((timer = timerInternals.removeNextEventTimer()) != null) {
- timers.add(timer);
- }
- while ((timer = timerInternals.removeNextProcessingTimer()) != null) {
- timers.add(timer);
- }
- while ((timer = timerInternals.removeNextSynchronizedProcessingTimer()) != null) {
- timers.add(timer);
- }
- if (timers.isEmpty()) {
- break;
- }
- reduceFnRunner.onTimers(timers);
- timers.clear();
+ TimerInternals.TimerData timer;
+ while ((timer = timerInternals.removeNextEventTimer()) != null) {
+ timers.add(timer);
+ }
+ while ((timer = timerInternals.removeNextProcessingTimer()) != null) {
+ timers.add(timer);
+ }
+ while ((timer = timerInternals.removeNextSynchronizedProcessingTimer()) != null) {
+ timers.add(timer);
+ }
+ if (timers.isEmpty()) {
+ break;
+ }
+ reduceFnRunner.onTimers(timers);
+ timers.clear();
}
}
}