<commit_before>/*
 * Copyright 2010 the original author or authors.
 * 
 * Licensed 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.openehealth.ipf.platform.camel.ihe.mllp.core.intercept.consumer;

import static org.openehealth.ipf.platform.camel.ihe.mllp.core.MllpMarshalUtils.appendSegments;
import static org.openehealth.ipf.platform.camel.ihe.mllp.core.MllpMarshalUtils.isEmpty;
import static org.openehealth.ipf.platform.camel.ihe.mllp.core.MllpMarshalUtils.isPresent;
import static org.openehealth.ipf.platform.camel.ihe.mllp.core.MllpMarshalUtils.joinSegments;
import static org.openehealth.ipf.platform.camel.ihe.mllp.core.MllpMarshalUtils.splitString;
import static org.openehealth.ipf.platform.camel.ihe.mllp.core.MllpMarshalUtils.uniqueId;

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

import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.openehealth.ipf.modules.hl7.message.MessageUtils;
import org.openehealth.ipf.modules.hl7dsl.MessageAdapter;
import org.openehealth.ipf.platform.camel.core.util.Exchanges;
import org.openehealth.ipf.platform.camel.ihe.mllp.core.ContinuationStorage;
import org.openehealth.ipf.platform.camel.ihe.mllp.core.MllpEndpoint;
import org.openehealth.ipf.platform.camel.ihe.mllp.core.MllpTransactionConfiguration;
import org.openehealth.ipf.platform.camel.ihe.mllp.core.intercept.AbstractMllpInterceptor;

import ca.uhn.hl7v2.model.Message;
import ca.uhn.hl7v2.parser.Parser;
import ca.uhn.hl7v2.util.Terser;


/**
 * Consumer-side interceptor for interactive continuation support 
 * as described in paragraph 5.6.3 of the HL7 v2.5 specification.
 * @author <NAME>
 */
public class ConsumerInteractiveResponseSenderInterceptor extends AbstractMllpInterceptor {
    private static final transient Log LOG = LogFactory.getLog(ConsumerInteractiveResponseSenderInterceptor.class);
    private final ContinuationStorage storage;
    
    public ConsumerInteractiveResponseSenderInterceptor(
            MllpEndpoint endpoint, 
            Processor wrappedProcessor,
            ContinuationStorage storage) 
    {
        super(endpoint, wrappedProcessor);
        this.storage = storage;
    }


    @Override
    public void process(Exchange exchange) throws Exception {
        Parser parser = getMllpEndpoint().getParser();
        MessageAdapter request = (MessageAdapter) exchange.getIn().getHeader(ORIGINAL_MESSAGE_ADAPTER_HEADER_NAME);
        Message requestMessage = (Message) request.getTarget();
        Terser requestTerser = new Terser(requestMessage);
        String requestMessageType = requestTerser.get("MSH-9-1");

        // get pieces of fragments' keys
        String msh31 = requestTerser.get("MSH-3-1");
        String msh32 = requestTerser.get("MSH-3-2");
        String msh33 = requestTerser.get("MSH-3-3");

        // handle cancel messages; if there is nothing to cancel -- pass to the route
        if ("QCN".equals(requestMessageType)) {
            String queryTag = requestTerser.get("QID-1");
            if (storage.deleteFragments(queryTag, msh31, msh32, msh33)) {
                LOG.debug("Dropped response chain for query ID " + queryTag);
                Message ack = MessageUtils.ack(parser.getFactory(), requestMessage);
                Exchanges.resultMessage(exchange).setBody(parser.encode(ack));
            } else {
                getWrappedProcessor().process(exchange);
            }
            return;
        }
        
        // check whether responses to messages of this type can even be splitted
        if (! getMllpEndpoint().getTransactionConfiguration().isContinuable(requestMessageType)) {
            getWrappedProcessor().process(exchange);
            return;
        }

        // check whether requested unit type is supported
        String rcp22 = requestTerser.get("RCP-2-2");
        if (!"RD".equals(rcp22)) {
            LOG.warn("Unit '" + rcp22 + "' in RCP-2-2 is not supported");
            getWrappedProcessor().process(exchange);
            return;
        }

        // determine the threshold (maximal records count per message)
        int threshold = -1;
        try {
            threshold = new Integer(requestTerser.get("RCP-2-1"));
        } catch (NumberFormatException nfe) {
            LOG.warn("Cannot parse RCP-2-1, try to use default threshold", nfe);
        }
        if (threshold < 1) {
            threshold = getMllpEndpoint().getInteractiveContinuationDefaultThreshold();
        }
        if (threshold < 1) {
            LOG.debug("Cannot perform interactive continuation: invalid or missing threshold");
            getWrappedProcessor().process(exchange);
            return;
        }
        
        // check whether the request is acceptable; if not -- pass it to the route, let the user decide 
        String continuationPointer = requestTerser.get("DSC-1");
        if (isEmpty(continuationPointer)) {
            continuationPointer = null;
        }

        if ((continuationPointer != null) && ! "I".equals(requestTerser.get("DSC-2"))) {
            LOG.warn("Cannot perform interactive continuation: DSC-1 is not empty and DSC-2 is not 'I'");
            getWrappedProcessor().process(exchange);
            return;
        }
        
        String queryTag = requestTerser.get("QPD-2");
        if (isEmpty(queryTag)) {
            LOG.warn("Cannot perform interactive continuation: empty query tag in QPD-2");
            getWrappedProcessor().process(exchange);
            return;
        }
<commit_msg>tiny improvement in parsing of integers<commit_after>/*
 * Copyright 2010 the original author or authors.
 * 
 * Licensed 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.openehealth.ipf.platform.camel.ihe.mllp.core.intercept.consumer;

import static org.openehealth.ipf.platform.camel.ihe.mllp.core.MllpMarshalUtils.appendSegments;
import static org.openehealth.ipf.platform.camel.ihe.mllp.core.MllpMarshalUtils.isEmpty;
import static org.openehealth.ipf.platform.camel.ihe.mllp.core.MllpMarshalUtils.isPresent;
import static org.openehealth.ipf.platform.camel.ihe.mllp.core.MllpMarshalUtils.joinSegments;
import static org.openehealth.ipf.platform.camel.ihe.mllp.core.MllpMarshalUtils.splitString;
import static org.openehealth.ipf.platform.camel.ihe.mllp.core.MllpMarshalUtils.uniqueId;

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

import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.openehealth.ipf.modules.hl7.message.MessageUtils;
import org.openehealth.ipf.modules.hl7dsl.MessageAdapter;
import org.openehealth.ipf.platform.camel.core.util.Exchanges;
import org.openehealth.ipf.platform.camel.ihe.mllp.core.ContinuationStorage;
import org.openehealth.ipf.platform.camel.ihe.mllp.core.MllpEndpoint;
import org.openehealth.ipf.platform.camel.ihe.mllp.core.MllpTransactionConfiguration;
import org.openehealth.ipf.platform.camel.ihe.mllp.core.intercept.AbstractMllpInterceptor;

import ca.uhn.hl7v2.model.Message;
import ca.uhn.hl7v2.parser.Parser;
import ca.uhn.hl7v2.util.Terser;


/**
 * Consumer-side interceptor for interactive continuation support 
 * as described in paragraph 5.6.3 of the HL7 v2.5 specification.
 * @author <NAME>
 */
public class ConsumerInteractiveResponseSenderInterceptor extends AbstractMllpInterceptor {
    private static final transient Log LOG = LogFactory.getLog(ConsumerInteractiveResponseSenderInterceptor.class);
    private final ContinuationStorage storage;
    
    public ConsumerInteractiveResponseSenderInterceptor(
            MllpEndpoint endpoint, 
            Processor wrappedProcessor,
            ContinuationStorage storage) 
    {
        super(endpoint, wrappedProcessor);
        this.storage = storage;
    }


    @Override
    public void process(Exchange exchange) throws Exception {
        Parser parser = getMllpEndpoint().getParser();
        MessageAdapter request = (MessageAdapter) exchange.getIn().getHeader(ORIGINAL_MESSAGE_ADAPTER_HEADER_NAME);
        Message requestMessage = (Message) request.getTarget();
        Terser requestTerser = new Terser(requestMessage);
        String requestMessageType = requestTerser.get("MSH-9-1");

        // get pieces of fragments' keys
        String msh31 = requestTerser.get("MSH-3-1");
        String msh32 = requestTerser.get("MSH-3-2");
        String msh33 = requestTerser.get("MSH-3-3");

        // handle cancel messages; if there is nothing to cancel -- pass to the route
        if ("QCN".equals(requestMessageType)) {
            String queryTag = requestTerser.get("QID-1");
            if (storage.deleteFragments(queryTag, msh31, msh32, msh33)) {
                LOG.debug("Dropped response chain for query ID " + queryTag);
                Message ack = MessageUtils.ack(parser.getFactory(), requestMessage);
                Exchanges.resultMessage(exchange).setBody(parser.encode(ack));
            } else {
                getWrappedProcessor().process(exchange);
            }
            return;
        }
        
        // check whether responses to messages of this type can even be splitted
        if (! getMllpEndpoint().getTransactionConfiguration().isContinuable(requestMessageType)) {
            getWrappedProcessor().process(exchange);
            return;
        }

        // check whether requested unit type is supported
        String rcp22 = requestTerser.get("RCP-2-2");
        if (!"RD".equals(rcp22)) {
            LOG.warn("Unit '" + rcp22 + "' in RCP-2-2 is not supported");
            getWrappedProcessor().process(exchange);
            return;
        }

        // determine the threshold (maximal records count per message)
        int threshold = -1;
        try {
            threshold = Integer.parseInt(requestTerser.get("RCP-2-1"));
        } catch (NumberFormatException nfe) {
            LOG.warn("Cannot parse RCP-2-1, try to use default threshold", nfe);
        }
        if (threshold < 1) {
            threshold = getMllpEndpoint().getInteractiveContinuationDefaultThreshold();
        }
        if (threshold < 1) {
            LOG.debug("Cannot perform interactive continuation: invalid or missing threshold");
            getWrappedProcessor().process(exchange);
            return;
        }
        
        // check whether the request is acceptable; if not -- pass it to the route, let the user decide 
        String continuationPointer = requestTerser.get("DSC-1");
        if (isEmpty(continuationPointer)) {
            continuationPointer = null;
        }

        if ((continuationPointer != null) && ! "I".equals(requestTerser.get("DSC-2"))) {
            LOG.warn("Cannot perform interactive continuation: DSC-1 is not empty and DSC-2 is not 'I'");
            getWrappedProcessor().process(exchange);
            return;
        }
        
        String queryTag = requestTerser.get("QPD-2");
        if (isEmpty(queryTag)) {
            LOG.warn("Cannot perform interactive continuation: empty query tag in QPD-2");
            getWrappedProcessor().process(exchange);
            return;
        }

{'fraction_non_alphanumeric': 0.09933774834437085, 'fraction_numerical': 0.010792249202845229, 'mean_word_length': 2.994774657086871, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 5, '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': '1470983', 'n_tokens_mistral': 3561, 'n_tokens_neox': 3445, 'n_words': 808}