datatodatabasesink.java

OBPM是一个开源

        insertBeanIntoDatabase(table, bean);

        if (_log.isDebugEnabled())
        {
            _log.debug("Inserted bean " + origIdentity);
        }

        if (_ensureFkOrder && _fkTables.contains(table))
        {
            Identity  newIdentity  = buildIdentityFromPKs(table, bean);
            ArrayList finishedObjs = new ArrayList();

            _identityMap.put(origIdentity, newIdentity);

            // we're doing multiple passes so that we can insert as much objects in
            // one go as possible
            ArrayList identitiesToCheck = new ArrayList();

            identitiesToCheck.add(origIdentity);
            while (!identitiesToCheck.isEmpty() && !_waitingObjects.isEmpty())
            {
                Identity curIdentity    = (Identity)identitiesToCheck.get(0);
                Identity curNewIdentity = (Identity)_identityMap.get(curIdentity);

                identitiesToCheck.remove(0);
                finishedObjs.clear();
                for (Iterator waitingObjIt = _waitingObjects.iterator(); waitingObjIt.hasNext();)
                {
                    WaitingObject waitingObj = (WaitingObject)waitingObjIt.next();
                    Identity      fkIdentity = waitingObj.removePendingFK(curIdentity);

                    if (fkIdentity != null)
                    {
                        updateFKColumns(waitingObj.getObject(), fkIdentity.getForeignKeyName(), curNewIdentity);
                    }
                    if (!waitingObj.hasPendingFKs())
                    {
                        waitingObjIt.remove();
                        // we defer handling of the finished objects to avoid concurrent modification exceptions
                        finishedObjs.add(waitingObj.getObject());
                    }
                }
                for (Iterator finishedObjIt = finishedObjs.iterator(); finishedObjIt.hasNext();)
                {
                    DynaBean finishedObj = (DynaBean)finishedObjIt.next();
                    Table    tableForObj = _model.getDynaClassFor(finishedObj).getTable();
                    Identity objIdentity = buildIdentityFromPKs(tableForObj, finishedObj);

                    insertBeanIntoDatabase(tableForObj, finishedObj);
                    
                    Identity newObjIdentity = buildIdentityFromPKs(tableForObj, finishedObj);

                    _identityMap.put(objIdentity, newObjIdentity);
                    identitiesToCheck.add(objIdentity);
                    if (_log.isDebugEnabled())
                    {
                        _log.debug("Inserted deferred row " + objIdentity);
                    }
                }
            }
        }
    }

    /**
     * Inserts the bean into the database or batch queue.
     * 
     * @param table The table
     * @param bean  The bean
     */
    private void insertBeanIntoDatabase(Table table, DynaBean bean) throws DataSinkException
    {
        if (_useBatchMode)
        {
            _batchQueue.add(bean);
            if (_batchQueue.size() >= _batchSize)
            {
                purgeBatchQueue();
            }
        }
        else
        {
            insertSingleBeanIntoDatabase(table, bean);
        }
    }

    /**
     * Purges the batch queue by inserting the objects into the database.
     */
    private void purgeBatchQueue() throws DataSinkException
    {
        if (!_batchQueue.isEmpty())
        {
            try
            {
                _platform.insert(_connection, _model, _batchQueue);
                if (!_connection.getAutoCommit())
                {
                    _connection.commit();
                }
                if (_log.isDebugEnabled())
                {
                    _log.debug("Inserted " + _batchQueue.size() + " rows in batch mode ");
                }
            }
            catch (Exception ex)
            {
                if (_haltOnErrors)
                {
                    _platform.returnConnection(_connection);
                    throw new DataSinkException(ex);
                }
                else
                {
                    _log.warn("Exception while inserting " + _batchQueue.size() + " rows via batch mode into the database", ex);
                }
            }
            _batchQueue.clear();
        }
    }
    
    /**
     * Directly inserts the given bean into the database.
     * 
     * @param table The table of the bean
     * @param bean  The bean
     */
    private void insertSingleBeanIntoDatabase(Table table, DynaBean bean) throws DataSinkException
    {
        try
        {
            boolean    needTwoStepInsert = false;
            ForeignKey selfRefFk         = null;

            if (!_platform.isIdentityOverrideOn() &&
                _tablesWithSelfIdentityReference.contains(table))
            {
                selfRefFk = table.getSelfReferencingForeignKey();

                // in case of a self-reference (fk points to the very row that we're inserting)
                // and (at least) one of the pk columns is an identity column, we first need
                // to insert the row with the fk columns set to null
                Identity pkIdentity = buildIdentityFromPKs(table, bean);
                Identity fkIdentity = buildIdentityFromFK(table, selfRefFk, bean);

                if (pkIdentity.equals(fkIdentity))
                {
                    if (_tablesWithRequiredSelfReference.contains(table))
                    {
                        throw new DataSinkException("Can only insert rows with fk pointing to themselves when all fk columns can be NULL (row pk is " + pkIdentity + ")");
                    }
                    else
                    {
                        needTwoStepInsert = true;
                    }
                }
            }

            if (needTwoStepInsert)
            {
                // we first insert the bean without the fk, then in the second step we update the bean
                // with the row with the identity pk values
                ArrayList fkValues = new ArrayList();

                for (int idx = 0; idx < selfRefFk.getReferenceCount(); idx++)
                {
                    String columnName = selfRefFk.getReference(idx).getLocalColumnName();

                    fkValues.add(bean.get(columnName));
                    bean.set(columnName, null);
                }
                _platform.insert(_connection, _model, bean);
                for (int idx = 0; idx < selfRefFk.getReferenceCount(); idx++)
                {
                    bean.set(selfRefFk.getReference(idx).getLocalColumnName(), fkValues.get(idx));
                }
                _platform.update(_connection, _model, bean);
            }
            else
            {
                _platform.insert(_connection, _model, bean);
            }
            if (!_connection.getAutoCommit())
            {
                _connection.commit();
            }
        }
        catch (Exception ex)
        {
            if (_haltOnErrors)
            {
                _platform.returnConnection(_connection);
                throw new DataSinkException(ex);
            }
            else
            {
                _log.warn("Exception while inserting a row into the database", ex);
            }
        }
    }
    
    /**
     * Returns the name of the given foreign key. If it has no name, then a temporary one
     * is generated from the names of the relevant tables and columns.
     *
     * @param owningTable    The table owning the fk
     * @param fk             The foreign key
     * @return The name
     */
    private String getFKName(Table owningTable, ForeignKey fk)
    {
        if ((fk.getName() != null) && (fk.getName().length() > 0))
        {
            return fk.getName();
        }
        else
        {
            StringBuffer result = new StringBuffer();

            result.append(owningTable.getName());
            result.append("[");
            for (int idx = 0; idx < fk.getReferenceCount(); idx++)
            {
                if (idx > 0)
                {
                    result.append(",");
                }
                result.append(fk.getReference(idx).getLocalColumnName());
            }
            result.append("]->");
            result.append(fk.getForeignTableName());
            result.append("[");
            for (int idx = 0; idx < fk.getReferenceCount(); idx++)
            {
                if (idx > 0)
                {
                    result.append(",");
                }
                result.append(fk.getReference(idx).getForeignColumnName());
            }
            result.append("]");
            return result.toString();
        }
    }
    
    /**
     * Builds an identity object from the primary keys of the specified table using the
     * column values of the supplied bean.
     * 
     * @param table The table
     * @param bean  The bean
     * @return The identity
     */
    private Identity buildIdentityFromPKs(Table table, DynaBean bean)
    {
        Identity identity  = new Identity(table);
        Column[] pkColumns = table.getPrimaryKeyColumns();

        for (int idx = 0; idx < pkColumns.length; idx++)
        {
            identity.setColumnValue(pkColumns[idx].getName(), bean.get(pkColumns[idx].getName()));
        }
        return identity;
    }

    /**
     * Builds an identity object for the specified foreign key using the foreignkey column values
     * of the supplied bean.
     * 
     * @param owningTable The table owning the foreign key
     * @param fk          The foreign key
     * @param bean        The bean
     * @return The identity
     */
    private Identity buildIdentityFromFK(Table owningTable, ForeignKey fk, DynaBean bean)
    {
        Identity identity = new Identity(fk.getForeignTable(), getFKName(owningTable, fk));

        for (int idx = 0; idx < fk.getReferenceCount(); idx++)
        {
            Reference reference = (Reference)fk.getReference(idx);
            Object    value     = bean.get(reference.getLocalColumnName());

            if (value == null)
            {
                return null;
            }
            identity.setColumnValue(reference.getForeignColumnName(), value);
        }
        return identity;
    }

    /**
     * Updates the values of the columns constituting the indicated foreign key with the values
     * of the given identity.
     * 
     * @param bean     The bean whose columns shall be updated
     * @param fkName   The name of the foreign key
     * @param identity The target identity
     */
    private void updateFKColumns(DynaBean bean, String fkName, Identity identity)
    {
        Table      sourceTable = ((SqlDynaClass)bean.getDynaClass()).getTable();
        Table      targetTable = identity.getTable();
        ForeignKey fk          = null;

        for (int idx = 0; idx < sourceTable.getForeignKeyCount(); idx++)
        {
            ForeignKey curFk = sourceTable.getForeignKey(idx);

            if (curFk.getForeignTableName().equalsIgnoreCase(targetTable.getName()))
            {
                if (fkName.equals(getFKName(sourceTable, curFk)))
                {
                    fk = curFk;
                    break;
                }
            }
        }
        if (fk != null)
        {
            for (int idx = 0; idx < fk.getReferenceCount(); idx++)
            {
                Reference curRef       = fk.getReference(idx);
                Column    sourceColumn = curRef.getLocalColumn();
                Column    targetColumn = curRef.getForeignColumn();

                bean.set(sourceColumn.getName(), identity.getColumnValue(targetColumn.getName()));
            }
        }
    }
}