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.ecm.platform.importer.mqueues.consumer; 020 021/** 022 * Keep state of a batch according to a batch policy. 023 * 024 * @since 9.1 025 */ 026public class BatchState { 027 private final BatchPolicy policy; 028 private int counter; 029 private long endMs; 030 031 public enum State {FILLING, FULL, TIMEOUT, LAST} 032 033 State state = State.FILLING; 034 035 public BatchState(BatchPolicy policy) { 036 this.policy = policy; 037 } 038 039 public void start() { 040 endMs = System.currentTimeMillis() + policy.getTimeThreshold().toMillis(); 041 counter = 0; 042 state = State.FILLING; 043 } 044 045 public State inc() { 046 if (state != State.FILLING) { 047 throw new IllegalStateException("Try to add an item to a batch in non filling state:" + state); 048 } 049 counter++; 050 return getState(); 051 } 052 053 public void force() { 054 state = State.FULL; 055 } 056 057 public void last() { 058 state = State.LAST; 059 } 060 061 public State getState() { 062 if (state != State.FILLING) { 063 return state; 064 } 065 if (counter >= policy.getCapacity()) { 066 state = State.FULL; 067 } else if (System.currentTimeMillis() > endMs) { 068 state = State.TIMEOUT; 069 } 070 return state; 071 } 072 073 public int getSize() { 074 return counter; 075 } 076 077}