<commit_before>/**
 *
 * 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.camel.component.jpa;

import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.impl.PollingConsumer;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import javax.persistence.EntityManager;
import javax.persistence.Query;
import javax.persistence.EntityTransaction;
import java.util.List;

/**
 * @version $Revision$
 */
public class JpaConsumer extends PollingConsumer<Exchange> {
    private static final transient Log log = LogFactory.getLog(JpaConsumer.class);

    private final JpaEndpoint endpoint;
    private final EntityManager entityManager;
    private QueryFactory queryFactory;
    private DeleteHandler<Object> deleteHandler;
    private EntityTransaction transaction;

    public JpaConsumer(JpaEndpoint endpoint, Processor<Exchange> processor, EntityManager entityManager) {
        super(endpoint, processor);
        this.endpoint = endpoint;
        this.entityManager = entityManager;
    }

    /**
     * Invoked whenever we should be polled
     */
    public synchronized void run() {
        log.debug("Starting to poll for new database entities to process");
        transaction = entityManager.getTransaction();
        transaction.begin();

        try {
            Query query = getQueryFactory().createQuery(this);
            configureParameters(query);
            List results = query.getResultList();
            for (Object result : results) {
                if (log.isDebugEnabled()) {
                    log.debug("Processing new entity: " + result);
                }
                // lets turn the result into an exchange and fire it into the processor
                Exchange exchange = createExchange(result);
                getProcessor().onExchange(exchange);
                getDeleteHandler().deleteObject(this, result);
            }

            transaction.commit();
            transaction = null;
        }
        catch (RuntimeException e) {
            log.warn("Caught: " + e, e);
            if (transaction != null) {
                transaction.rollback();
            }
        }
    }

    // Properties
    //-------------------------------------------------------------------------

    public EntityManager getEntityManager() {
        return entityManager;
    }

    public JpaEndpoint getEndpoint() {
        return endpoint;
    }

    public QueryFactory getQueryFactory() {
        if (queryFactory == null) {
            queryFactory = createQueryFactory();
        }
        return queryFactory;
    }

    public void setQueryFactory(QueryFactory queryFactory) {
        this.queryFactory = queryFactory;
    }

    public DeleteHandler getDeleteHandler() {
        if (deleteHandler == null) {
            deleteHandler = createDeleteHandler();
        }
        return deleteHandler;
    }

    public void setDeleteHandler(DeleteHandler deleteHandler) {
        this.deleteHandler = deleteHandler;
    }

    // Implementation methods
    //-------------------------------------------------------------------------
    @Override
    protected synchronized void doStop() throws Exception {
        if (transaction != null) {
            transaction.rollback();
        }
        entityManager.close();
        super.doStop();
    }

    protected QueryFactory createQueryFactory() {
        Class<?> entityType = endpoint.getEntityType();
        if (entityType == null) {
            return null;
        }
        else {
            return QueryBuilder.query("select x from " + entityType.getName() + " x");
        }
    }

    protected DeleteHandler<Object> createDeleteHandler() {<commit_msg>use explicit exclusive lock when processing component<commit_after>/**
 *
 * 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.camel.component.jpa;

import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.impl.PollingConsumer;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import javax.persistence.EntityManager;
import javax.persistence.EntityTransaction;
import javax.persistence.LockModeType;
import javax.persistence.Query;
import java.util.List;

/**
 * @version $Revision$
 */
public class JpaConsumer extends PollingConsumer<Exchange> {
    private static final transient Log log = LogFactory.getLog(JpaConsumer.class);
    private final JpaEndpoint endpoint;
    private final EntityManager entityManager;
    private QueryFactory queryFactory;
    private DeleteHandler<Object> deleteHandler;
    private EntityTransaction transaction;

    public JpaConsumer(JpaEndpoint endpoint, Processor<Exchange> processor, EntityManager entityManager) {
        super(endpoint, processor);
        this.endpoint = endpoint;
        this.entityManager = entityManager;
    }

    /**
     * Invoked whenever we should be polled
     */
    public synchronized void run() {
        log.debug("Starting to poll for new database entities to process");
        transaction = entityManager.getTransaction();
        transaction.begin();

        try {
            Query query = getQueryFactory().createQuery(this);
            configureParameters(query);
            List results = query.getResultList();
            for (Object result : results) {
                if (log.isDebugEnabled()) {
                    log.debug("Processing new entity: " + result);
                }

                if (lockEntity(result)) {
                    // lets turn the result into an exchange and fire it into the processor
                    Exchange exchange = createExchange(result);
                    getProcessor().onExchange(exchange);
                    getDeleteHandler().deleteObject(this, result);
                }
            }

            transaction.commit();
            transaction = null;
        }
        catch (RuntimeException e) {
            log.warn("Caught: " + e, e);
            if (transaction != null) {
                transaction.rollback();
            }
        }
    }

    // Properties
    //-------------------------------------------------------------------------

    public EntityManager getEntityManager() {
        return entityManager;
    }

    public JpaEndpoint getEndpoint() {
        return endpoint;
    }

    public QueryFactory getQueryFactory() {
        if (queryFactory == null) {
            queryFactory = createQueryFactory();
        }
        return queryFactory;
    }

    public void setQueryFactory(QueryFactory queryFactory) {
        this.queryFactory = queryFactory;
    }

    public DeleteHandler getDeleteHandler() {
        if (deleteHandler == null) {
            deleteHandler = createDeleteHandler();
        }
        return deleteHandler;
    }

    public void setDeleteHandler(DeleteHandler deleteHandler) {
        this.deleteHandler = deleteHandler;
    }

    // Implementation methods
    //-------------------------------------------------------------------------
    @Override
    protected synchronized void doStop() throws Exception {
        if (transaction != null) {
            transaction.rollback();
        }
        entityManager.close();
        super.doStop();
    }

    /**
     * A strategy method to lock an object with an exclusive lock so that it can be processed
     *
     * @param entity the entity to be locked
     * @return true if the entity was locked
     */
    protected boolean lockEntity(Object entity) {
        try {
            if (log.isDebugEnabled()) {
                log.debug("Acquiring exclusive lock on entity: " + entity);
            }
            entityManager.lock(entity, LockModeType.WRITE);
            return true;
        }
        catch (Exception e) {
            if (log.isDebugEnabled()) {
                log.debug("Failed to achieve lock on entity: " + entity + ". Reason: " + e, e);
            }
            return false;
        }
    }

    protected QueryFactory createQueryFactory() {
        Class<?> entityType = endpoint.getEntityType();
        if (entityType == null) {
            return null;
        }
        else {
            return QueryBuilder.query("select x from " + entityType.getName() + " x");
        }
    }

    protected DeleteHandler<Object> createDeleteHandler() {
{'fraction_non_alphanumeric': 0.11197969543147208, 'fraction_numerical': 0.0008121827411167512, 'mean_word_length': 2.17671718800387, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7591339', 'n_tokens_mistral': 2326, 'n_tokens_neox': 2258, 'n_words': 731}