001/* 002 * (C) Copyright 2017 Nuxeo SA (http://nuxeo.com/) and others. 003 * 004 * Licensed under the Apache License, Version 2.0 (the "License"); 005 * you may not use this file except in compliance with the License. 006 * You may obtain a copy of the License at 007 * 008 * http://www.apache.org/licenses/LICENSE-2.0 009 * 010 * Unless required by applicable law or agreed to in writing, software 011 * distributed under the License is distributed on an "AS IS" BASIS, 012 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 013 * See the License for the specific language governing permissions and 014 * limitations under the License. 015 * 016 * Contributors: 017 * bdelbosc 018 */ 019package org.nuxeo.lib.stream.pattern.consumer.internals; 020 021import org.nuxeo.lib.stream.pattern.consumer.BatchPolicy; 022 023/** 024 * Keep state of a batch according to a batch policy. 025 * 026 * @since 9.1 027 */ 028public class BatchState { 029 protected final BatchPolicy policy; 030 031 protected int counter; 032 033 protected long endMs; 034 035 public enum State { 036 FILLING, FULL, TIMEOUT, LAST 037 } 038 039 State state = State.FILLING; 040 041 public BatchState(BatchPolicy policy) { 042 this.policy = policy; 043 } 044 045 public void start() { 046 endMs = System.currentTimeMillis() + policy.getTimeThreshold().toMillis(); 047 counter = 0; 048 state = State.FILLING; 049 } 050 051 public State inc() { 052 if (state != State.FILLING) { 053 throw new IllegalStateException("Try to add an item to a batch in non filling state:" + state); 054 } 055 counter++; 056 return getState(); 057 } 058 059 public void force() { 060 state = State.FULL; 061 } 062 063 public void last() { 064 state = State.LAST; 065 } 066 067 public State getState() { 068 if (state != State.FILLING) { 069 return state; 070 } 071 if (counter >= policy.getCapacity()) { 072 state = State.FULL; 073 } else if (System.currentTimeMillis() > endMs) { 074 state = State.TIMEOUT; 075 } 076 return state; 077 } 078 079 public int getSize() { 080 return counter; 081 } 082 083}