class StockOp {

    @CoGroup
    public void cutoff(
            // 在庫は商品ごとにグルーピングして、さらに入荷順に並べる
            @Key(group = "ITEM_CODE", order = "PURCHASED_DATE ASC") List<Stock> stocks,
            // 注文も商品ごとにグルーピングして、さらに出荷順に並べる
            @Key(group = "ITEM_CODE", order = "SHIPPED_DATE ASC") List<Shipment> shipments,
            Result<Stock> newStocks,
            Result<Shipment> newShipments) {
        Iterator<Stock> eachStock = stocks.iterator();
        Iterator<Shipment> eachShipment = shipments.iterator();
        
        // 在庫が空になるまで
        while (eachStock.hasNext()) {
            Stock stock = eachStock.next();
            
            // 注文を処理しきるまで
            while (eachShipment.hasNext()) {
                // この在庫レコードが終わったら次へ
                if (stock.getQuantity() == 0) {
                    break;
                }
                // 注文をひとつ取り出して処理
                Shipment shipment = eachShipment.next();
                shipment.setCost(stock.getCost());
                newShipments.add(shipment);
                
                // 在庫を減らす
                stock.getQuantityOption().add(-1);
            }
            
            // 現在の在庫情報を出力
            newStocks.add(stock);
        }
        
        // 出荷したはずが在庫が足りてないですよ...
        // (書き戻さなくてもよく、その場合は更新されないだけ)
        while (eachShipment.hasNext()) {
            Shipment shipment = eachShipment.next();
            newShipments.add(shipment);
        }
    }

}

class StockOpTest {

    @Test
    public void testCutoff() {

        StockOpImpl operator = new StockOpImpl();

        List<Stock> stocks = new ArrayList<Stock>();
        stocks.add(StockFactory.create(new DateTime(), 0, 100, 10));

        List<Shipment> shipments = new ArrayList<Shipment>();
        shipments.add(ShipmentFactory.create(new DateTime(), 10, 100));

        MockResult<Stock> newStocks = new MockResult<Stock>();
        MockResult<Shipment> newShipments = new MockResult<Shipment>();

        operator.cutoff(stocks, shipments, newStocks, newShipments);

        assertThat(newStocks.getResults().size(), is(1));
        assertThat(newShipments.getResults().size(), is(1));
    }

}