largeselect.java

另外一种持久性o/m软件

    /**
     * Retrieve a specific page, if it exists.
     *
     * @param pageNumber the number of the page to be retrieved - must be
     * greater than zero.  An empty <code>List</code> will be returned if
     * <code>pageNumber</code> exceeds the total number of pages that exist.
     * @return a <code>List</code> of query results containing a maximum of
     * <code>pageSize</code> results.
     * @throws IllegalArgumentException when <code>pageNo</code> is not
     * greater than zero.
     * @throws TorqueException if invoking the <code>populateObjects()<code>
     * method runs into problems or a sleep is unexpectedly interrupted.
     */
    public List getPage(int pageNumber) throws TorqueException
    {
        if (pageNumber < 1)
        {
            throw new IllegalArgumentException("pageNumber must be greater "
                    + "than zero.");
        }
        currentPageNumber = pageNumber;
        return getResults((pageNumber - 1) * pageSize);
    }

    /**
     * Gets the next page of rows.
     *
     * @return a <code>List</code> of query results containing a maximum of
     * <code>pageSize</code> reslts.
     * @throws TorqueException if invoking the <code>populateObjects()<code>
     * method runs into problems or a sleep is unexpectedly interrupted.
     */
    public List getNextResults() throws TorqueException
    {
        if (!getNextResultsAvailable())
        {
            return getCurrentPageResults();
        }
        currentPageNumber++;
        return getResults(position);
    }

    /**
     * Provide access to the results from the current page.
     *
     * @return a <code>List</code> of query results containing a maximum of
     * <code>pageSize</code> reslts.
     */
    public List getCurrentPageResults()
    {
        return lastResults;
    }

    /**
     * Gets the previous page of rows.
     *
     * @return a <code>List</code> of query results containing a maximum of
     * <code>pageSize</code> reslts.
     * @throws TorqueException if invoking the <code>populateObjects()<code>
     * method runs into problems or a sleep is unexpectedly interrupted.
     */
    public List getPreviousResults() throws TorqueException
    {
        if (!getPreviousResultsAvailable())
        {
            return getCurrentPageResults();
        }

        int start;
        if (position - 2 * pageSize < 0)
        {
            start = 0;
            currentPageNumber = 1;
        }
        else
        {
            start = position - 2 * pageSize;
            currentPageNumber--;
        }
        return getResults(start);
    }

    /**
     * Gets a page of rows starting at a specified row.
     *
     * @param start the starting row.
     * @return a <code>List</code> of query results containing a maximum of
     * <code>pageSize</code> reslts.
     * @throws TorqueException if invoking the <code>populateObjects()<code>
     * method runs into problems or a sleep is unexpectedly interrupted.
     */
    private List getResults(int start) throws TorqueException
    {
        return getResults(start, pageSize);
    }

    /**
     * Gets a block of rows starting at a specified row and containing a
     * specified number of rows.
     *
     * @param start the starting row.
     * @param size the number of rows.
     * @return a <code>List</code> of query results containing a maximum of
     * <code>pageSize</code> reslts.
     * @throws IllegalArgumentException if <code>size &gt; memoryLimit</code> or
     * <code>start</code> and <code>size</code> result in a situation that is
     * not catered for.
     * @throws TorqueException if invoking the <code>populateObjects()<code>
     * method runs into problems or a sleep is unexpectedly interrupted.
     */
    private synchronized List getResults(int start, int size)
            throws IllegalArgumentException, TorqueException
    {
        if (log.isDebugEnabled())
        {
            log.debug("getResults(start: " + start
                    + ", size: " + size + ") invoked.");
        }

        if (size > memoryLimit)
        {
            throw new IllegalArgumentException("size (" + size
                    + ") exceeds memory limit (" + memoryLimit + ").");
        }

        // Request was for a block of rows which should be in progess.
        // If the rows have not yet been returned, wait for them to be
        // retrieved.
        if (start >= blockBegin && (start + size - 1) <= blockEnd)
        {
            if (log.isDebugEnabled())
            {
                log.debug("getResults(): Sleeping until "
                        + "start+size-1 (" + (start + size - 1)
                        + ") > currentlyFilledTo (" + currentlyFilledTo
                        + ") && !queryCompleted (!" + queryCompleted + ")");
            }
            while (((start + size - 1) > currentlyFilledTo) && !queryCompleted)
            {
                try
                {
                    Thread.sleep(500);
                }
                catch (InterruptedException e)
                {
                    throw new TorqueException("Unexpected interruption", e);
                }
            }
        }

        // Going in reverse direction, trying to limit db hits so assume user
        // might want at least 2 sets of data.
        else if (start < blockBegin && start >= 0)
        {
            if (log.isDebugEnabled())
            {
                log.debug("getResults(): Paging backwards as start (" + start
                        + ") < blockBegin (" + blockBegin + ") && start >= 0");
            }
            stopQuery();
            if (memoryLimit >= 2 * size)
            {
                blockBegin = start - size;
                if (blockBegin < 0)
                {
                    blockBegin = 0;
                }
            }
            else
            {
                blockBegin = start;
            }
            blockEnd = blockBegin + memoryLimit - 1;
            startQuery(size);
            // Re-invoke getResults() to provide the wait processing.
            return getResults(start, size);
        }

        // Assume we are moving on, do not retrieve any records prior to start.
        else if ((start + size - 1) > blockEnd)
        {
            if (log.isDebugEnabled())
            {
                log.debug("getResults(): Paging past end of loaded data as "
                        + "start+size-1 (" + (start + size - 1)
                        + ") > blockEnd (" + blockEnd + ")");
            }
            stopQuery();
            blockBegin = start;
            blockEnd = blockBegin + memoryLimit - 1;
            startQuery(size);
            // Re-invoke getResults() to provide the wait processing.
            return getResults(start, size);
        }

        else
        {
            throw new IllegalArgumentException("Parameter configuration not "
                    + "accounted for.");
        }

        int fromIndex = start - blockBegin;
        int toIndex = fromIndex + Math.min(size, results.size() - fromIndex);

        if (log.isDebugEnabled())
        {
            log.debug("getResults(): Retrieving records from results elements "
                    + "start-blockBegin (" + fromIndex + ") through "
                    + "fromIndex + Math.min(size, results.size() - fromIndex) ("
                    + toIndex + ")");
        }

        List returnResults;

        synchronized (results) 
        {
            returnResults = new ArrayList(results.subList(fromIndex, toIndex));
        }

        if (null != returnBuilderClass)
        {
            // Invoke the populateObjects() method
            Object[] theArgs = { returnResults };
            try
            {
                returnResults =
                    (List) populateObjectsMethod.invoke(
                        returnBuilderClass.newInstance(),
                        theArgs);
            }
            catch (Exception e)
            {
                throw new TorqueException("Unable to populate results", e);
            }
        }
        position = start + size;
        lastResults = returnResults;
        return returnResults;
    }

    /**
     * A background thread that retrieves the rows.
     */
    public void run()
    {
        int size = pageSize;
        /* The connection to the database. */
        Connection conn = null;

        try
        {
            // Add 1 to memory limit to check if the query ends on a page break.
            results = new ArrayList(memoryLimit + 1);

            // Use the criteria to limit the rows that are retrieved to the
            // block of records that fit in the predefined memoryLimit.
            criteria.setOffset(blockBegin);
            // Add 1 to memory limit to check if the query ends on a page break.
            criteria.setLimit(memoryLimit + 1);
            query = BasePeer.createQueryString(criteria);

            // Get a connection to the db.
            conn = Torque.getConnection(dbName);

            // Execute the query.
            if (log.isDebugEnabled())
            {
                log.debug("run(): query = " + query);
                log.debug("run(): memoryLimit = " + memoryLimit);
                log.debug("run(): blockBegin = " + blockBegin);
                log.debug("run(): blockEnd = " + blockEnd);
            }
            qds = new QueryDataSet(conn, query);

            // Continue getting rows one page at a time until the memory limit
            // is reached, all results have been retrieved, or the rest
            // of the results have been determined to be irrelevant.
            while (!killThread
                && !qds.allRecordsRetrieved()
                && currentlyFilledTo + pageSize <= blockEnd)
            {
                // This caters for when memoryLimit is not a multiple of
                //  pageSize which it never is because we always add 1 above.
                if ((currentlyFilledTo + pageSize) >= blockEnd)
                {
                    // Add 1 to check if the query ends on a page break.
                    size = blockEnd - currentlyFilledTo + 1;
                }

                if (log.isDebugEnabled())
                {
                    log.debug("run(): Invoking BasePeer.getSelectResults(qds, "
                            + size + ", false)");
                }

                List tempResults
                        = BasePeer.getSelectResults(qds, size, false);

                synchronized (results) 
                {
                    for (int i = 0, n = tempResults.size(); i < n; i++)
                    {
                        results.add(tempResults.get(i));
                    }
                }

                currentlyFilledTo += tempResults.size();
                boolean perhapsLastPage = true;

                // If the extra record was indeed found then we know we are not
                // on the last page but we must now get rid of it.
                if (results.size() == memoryLimit + 1)
                {
                    synchronized (results)
                    {
                        results.remove(currentlyFilledTo--);
                    }
                    perhapsLastPage = false;
                }

                if (results.size() > 0
                    && blockBegin + currentlyFilledTo >= totalRecords)
                {
                    // Add 1 because index starts at 0
                    totalRecords = blockBegin + currentlyFilledTo + 1;
                }

                if (qds.allRecordsRetrieved())
                {
                    queryCompleted = true;
                    // The following ugly condition ensures that the totals are
                    // not finalized when a user does something like requesting
                    // a page greater than what exists in the database.
                    if (perhapsLastPage
                        && getCurrentPageNumber() <= getTotalPages())
                    {
                        totalsFinalized = true;
                    }
                }
                qds.clearRecords();
            }

            if (log.isDebugEnabled())
            {
                log.debug("run(): While loop terminated because either:");
                log.debug("run(): 1. qds.allRecordsRetrieved(): "
                        + qds.allRecordsRetrieved());
                log.debug("run(): 2. killThread: " + killThread);
                log.debug("run(): 3. !(currentlyFilledTo + size <= blockEnd): !"
                        + (currentlyFilledTo + pageSize <= blockEnd));
                log.debug("run(): - currentlyFilledTo: " + currentlyFilledTo);
                log.debug("run(): - size: " + pageSize);
                log.debug("run(): - blockEnd: " + blockEnd);
                log.debug("run(): - results.size(): " + results.size());
            }
        }
        catch (TorqueException e)
        {
            log.error(e);
        }
        catch (SQLException e)
        {
            log.error(e);
        }
        catch (DataSetException e)
        {
            log.error(e);
        }
        finally
        {
            try
            {
                if (qds != null)
                {
                    qds.close();
                }
                Torque.closeConnection(conn);
            }
            catch (SQLException e)
            {
                log.error(e);
            }
            catch (DataSetException e)
            {
                log.error(e);
            }
            threadRunning = false;
        }
    }

    /**
     * Starts a new thread to retrieve the result set.
     *
     * @param initialSize the initial size for each block.
     */
    private synchronized void startQuery(int initialSize)
    {
        if (!threadRunning)
        {
            pageSize = initialSize;
            currentlyFilledTo = -1;
            queryCompleted = false;
            thread = new Thread(this);
            thread.start();
            threadRunning = true;
        }
    }

    /**
     * Used to stop filling the memory with the current block of results, if it
     * has been determined that they are no longer relevant.
     *
     * @throws TorqueException if a sleep is interrupted.
     */
    private synchronized void stopQuery() throws TorqueException