package org.flymine.dataconversion;

/*
 * Copyright (C) 2002-2003 FlyMine
 *
 * This code may be freely distributed and modified under the
 * terms of the GNU Lesser General Public Licence.  This should
 * be distributed with the code.  See the LICENSE file for more
 * information or http://www.gnu.org/copyleft/lesser.html.
 *
 */

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;

import org.flymine.model.fulldata.Item;
import org.flymine.objectstore.ObjectStoreException;
import org.flymine.util.ObjectPipe;

import org.apache.log4j.Logger;

/**
 * Stores Items in another ItemStore. This class decouples the thread that calls methods in this
 * class from the actual act of storing the Items, improving performance. Actual storing is done in
 * a separate Thread. This class uses an ObjectPipe to communicate the Items to be stored to the
 * Thread that performs the store. The store() and storeAll() methods should never take much time to
 * return to the caller unless the buffer is full, however the close() method will block until all
 * the Items are fully stored, or until an exception is thrown.
 *
 * @author Matthew Wakeling
 */
public class BufferedItemWriter implements ItemWriter
{
    protected static final Logger LOG = Logger.getLogger(BufferedItemWriter.class);
    private ObjectPipe pipe = new ObjectPipe(PIPE_LENGTH);
    private ItemWriter iw;
    private int batchCounter = 0;
    private List batch = new ArrayList();
    private ObjectStoreException problem = null;
    private boolean finished = false;
    private static final int BATCH_SIZE = 1000;
    private static final int PIPE_LENGTH = 5;
    
    /**
     * Constructs the ItemWriter with another ItemWriter.
     *
     * @param iw the ItemWriter in which to store the Items
     */
    public BufferedItemWriter(ItemWriter iw) {
        this.iw = iw;
        PipeReader pipeReader = new PipeReader();
        Thread pipeReaderThread = new Thread(pipeReader, "BufferedItemWriter.PipeReader thread");
        pipeReaderThread.start();
    }

    /**
     * @see ItemWriter#store
     */
    public void store(Item item) throws ObjectStoreException {
        batch.add(item);
        batchCounter++;
        if (batchCounter >= BATCH_SIZE) {
            pipe.put(batch);
            batch = new ArrayList();
            batchCounter = 0;
            checkException();
        }
    }

    /**
     * @see ItemWriter#storeAll
     */
    public void storeAll(Collection items) throws ObjectStoreException {
        if (batchCounter > 0) {
            pipe.put(batch);
            batch = new ArrayList();
            batchCounter = 0;
        }
        pipe.put(batch);
        checkException();
    }

    
    /**
     * @see ItemWriter#close
     */
    public void close() throws ObjectStoreException {
        if (batchCounter > 0) {
            pipe.put(batch);
            batchCounter = 0;
        }
        pipe.finish();
        synchronized (this) {
            while (!finished) {
                try {
                    wait();
                } catch (InterruptedException e) {
                }
            }
        }
        checkException();
    }

    private synchronized void checkException() throws ObjectStoreException {
        if (problem != null) {
            throw problem;
        }
    }

    private class PipeReader implements Runnable
    {
        public PipeReader() {
        }
        
        public void run() {
            while (pipe.hasNext()) {
                Object nextInPipe = pipe.next();
                if (problem == null) {
                    try {
                        Collection col = (Collection) nextInPipe;
                        iw.storeAll(col);
                    } catch (ObjectStoreException e) {
                        synchronized (BufferedItemWriter.this) {
                            problem = e;
                        }
                    } catch (Exception e) {
                        synchronized (BufferedItemWriter.this) {
                            problem = new ObjectStoreException(e);
                        }
                    }
                }
            }
            try {
                iw.close();
            } catch (ObjectStoreException e) {
                synchronized (BufferedItemWriter.this) {
                    if (problem == null) {
                        problem = e;
                    }
                }
            }
            synchronized (BufferedItemWriter.this) {
                finished = true;
                BufferedItemWriter.this.notifyAll();
            }
        }
    }
}

@@ -381,24 +381,51 @@
 Collection;%0A
+import java.util.Iterator;%0A
 import java.
@@ -470,20 +470,114 @@
 ulldata.
-Item
+Attribute;%0Aimport org.flymine.model.fulldata.Item;%0Aimport org.flymine.model.fulldata.ReferenceList
 ;%0Aimport
@@ -1546,24 +1546,62 @@
 ounter = 0;%0A
+    private int batchCharCounter = 0;%0A
     private 
@@ -1765,16 +1765,73 @@
 = 1000;%0A
+    private static final int BATCH_CHAR_SIZE = 10000000;%0A
     priv
@@ -1869,9 +1869,9 @@
 H = 
-5
+3
 ;%0A  
@@ -2372,32 +2372,463 @@
 toreException %7B%0A
+        int itemSize = 100;%0A        Iterator iter = item.getAttributes().iterator();%0A        while (iter.hasNext()) %7B%0A            itemSize += ((Attribute) iter.next()).getValue().length() + 50;%0A        %7D%0A        iter = item.getCollections().iterator();%0A        while (iter.hasNext()) %7B%0A            itemSize += ((ReferenceList) iter.next()).getRefIds().length() + 50;%0A        %7D%0A        itemSize += item.getReferences().size() * 50;%0A
         batch.ad
@@ -2860,16 +2860,54 @@
 nter++;%0A
+        batchCharCounter += itemSize;%0A
         
@@ -2901,32 +2901,33 @@
 ize;%0A        if 
+(
 (batchCounter %3E=
@@ -2938,16 +2938,58 @@
 CH_SIZE)
+ %7C%7C (batchCharCounter %3E= BATCH_CHAR_SIZE))
  %7B%0A     
@@ -3071,32 +3071,66 @@
 tchCounter = 0;%0A
+            batchCharCounter = 0;%0A
             chec
@@ -3401,32 +3401,66 @@
 tchCounter = 0;%0A
+            batchCharCounter = 0;%0A
         %7D%0A      
@@ -3698,32 +3698,66 @@
 tchCounter = 0;%0A
+            batchCharCounter = 0;%0A
         %7D%0A      

Fix for when huge Items turn up