bpel-runtime/src/main/java/org/apache/ode/bpel/engine/ProcessCleanUpRunnable.java

/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
package org.apache.ode.bpel.engine;

import java.io.Serializable;
import java.util.Map;
import java.util.concurrent.Callable;

import javax.xml.namespace.QName;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.ode.bpel.dao.DeferredProcessInstanceCleanable;
import org.apache.ode.bpel.dao.ProcessDAO;
import org.apache.ode.bpel.engine.BpelServerImpl.ContextsAware;
import org.apache.ode.bpel.iapi.Scheduler.JobDetails;
import org.apache.ode.bpel.iapi.Scheduler.MapSerializableRunnable;

public class ProcessCleanUpRunnable implements MapSerializableRunnable, ContextsAware {
    private static final long serialVersionUID = 1L;

    private static final Logger __log = LoggerFactory.getLogger(ProcessCleanUpRunnable.class);

    public final static int PROCESS_CLEANUP_TRANSACTION_SIZE = Integer.getInteger("org.apache.ode.processInstanceDeletion.transactionSize", 10);

    private transient Contexts _contexts;
    private transient Long _pidId;

    public ProcessCleanUpRunnable() {
    }
    
    public ProcessCleanUpRunnable(Long pidId) {
        _pidId = pidId;
    }

    public void storeToDetails(JobDetails details) {
        details.getDetailsExt().put("pidId", _pidId);
    }

    public void restoreFromDetails(JobDetails details) {
        _pidId = (Long) details.getDetailsExt().get("pidId");
    }

    public void setContexts(Contexts contexts) {
        _contexts = contexts;
    }

    public void run() {
        if(__log.isDebugEnabled()) __log.debug("Deleting runtime data for old process: " + _pidId + "...");
        try {
            // deleting of a process may involve hours' of database transaction,
            // we need to break it down to smaller transactions
            int transactionResultSize = 0;
            do {
                transactionResultSize = _contexts.scheduler.execTransaction(new Callable<Integer>() {
                    public Integer call() throws Exception {
                        ProcessDAO process = _contexts.dao.getConnection().createTransientProcess(_pidId);
                        if( !(process instanceof DeferredProcessInstanceCleanable) ) {
                            throw new IllegalArgumentException("ProcessDAO does not implement DeferredProcessInstanceCleanable!!!");
                        }
                        return ((DeferredProcessInstanceCleanable)process).deleteInstances(PROCESS_CLEANUP_TRANSACTION_SIZE);
                    }
                });
                if(__log.isDebugEnabled()) __log.debug("Deleted " + transactionResultSize + "instances for old process: " + _pidId + ".");
            } while( transactionResultSize == PROCESS_CLEANUP_TRANSACTION_SIZE );
        } catch (RuntimeException re) {
            throw re;
        } catch (Exception ex) {
            throw new RuntimeException(ex);
        }
        if(__log.isInfoEnabled()) __log.info("Deleted runtime data for old process: " + _pidId + ".");
    }
}