import org.apache.synapse.MessageContext; //导入依赖的package包/类
/**
 * @param remoteFile     Location of the file
 * @param destination    New file location
 * @param filePattern    Pattern of the file
 * @param manager        Standard file system manager
 * @param messageContext The message context that is generated for processing the file
 * @throws IOException
 */
private void moveFileWithPattern(FileObject remoteFile, String destination, String filePattern,
                                 StandardFileSystemManager manager, MessageContext messageContext) throws IOException {
    FilePattenMatcher patternMatcher = new FilePattenMatcher(filePattern);
    try {
        if (patternMatcher.validate(remoteFile.getName().getBaseName())) {
            FileObject file = manager.resolveFile(destination, FileConnectorUtils.init(messageContext));
            if (!file.exists()) {
                file.createFolder();
            }
            file = manager.resolveFile(destination + File.separator + remoteFile.getName().getBaseName(),
                    FileConnectorUtils.init(messageContext));
            remoteFile.moveTo(file);
        }
    } catch (IOException e) {
        log.error("Error occurred while moving a file. " + e.getMessage(), e);
    }
}
 

import org.apache.synapse.MessageContext; //导入依赖的package包/类
/**
 * Get the correct formatter for message
 *
 * @param msgContext The message context that is generated for processing the file
 */
private MessageFormatter getMessageFormatter(org.apache.axis2.context.MessageContext msgContext) {
    OMElement firstChild = msgContext.getEnvelope().getBody().getFirstElement();
    if (firstChild != null) {
        if (BaseConstants.DEFAULT_BINARY_WRAPPER.equals(firstChild.getQName())) {
            return new BinaryFormatter();
        } else if (BaseConstants.DEFAULT_TEXT_WRAPPER.equals(firstChild.getQName())) {
            return new PlainTextFormatter();
        }
    }
    try {
        return MessageProcessorSelector.getMessageFormatter(msgContext);
    } catch (Exception e) {
        throw new BaseTransportException("Unable to get the message formatter to use");
    }
}
 

import org.apache.synapse.MessageContext; //导入依赖的package包/类
public void connect(MessageContext messageContext) {
    boolean includeParentDirectory;
    String source = (String) ConnectorUtils.lookupTemplateParamater(messageContext,
            FileConstants.FILE_LOCATION);
    String destination = (String) ConnectorUtils.lookupTemplateParamater(messageContext,
            FileConstants.NEW_FILE_LOCATION);
    String filePattern = (String) ConnectorUtils.lookupTemplateParamater(messageContext,
            FileConstants.FILE_PATTERN);
    String includeParentDir = (String) ConnectorUtils.lookupTemplateParamater(messageContext,
            FileConstants.INCLUDE_PARENT_DIRECTORY);
    if (StringUtils.isEmpty(includeParentDir)) {
        includeParentDirectory = Boolean.parseBoolean(FileConstants.DEFAULT_INCLUDE_PARENT_DIRECTORY);
    } else {
        includeParentDirectory = Boolean.parseBoolean(includeParentDir);
    }
    FileSystemOptions opts = FileConnectorUtils.init(messageContext);
    boolean resultStatus = copyFile(source, destination, filePattern, messageContext, opts, includeParentDirectory);
    ResultPayloadCreate resultPayload = new ResultPayloadCreate();
    generateResults(messageContext, resultStatus, resultPayload);
}
 

import org.apache.synapse.MessageContext; //导入依赖的package包/类
public void connect(MessageContext messageContext) {
    String source = (String) ConnectorUtils.lookupTemplateParamater(messageContext,
            FileConstants.FILE_LOCATION);
    String destination = (String) ConnectorUtils.lookupTemplateParamater(messageContext,
            FileConstants.NEW_FILE_LOCATION);
    boolean resultStatus = false;
    try {
        resultStatus = new FileUnzipUtil().unzip(source, destination, messageContext);
    } catch (Exception e) {
        handleException(e.getMessage(), messageContext);
    }
    generateResults(messageContext, resultStatus);
    if (log.isDebugEnabled()) {
        log.debug("File extracted to" + destination);
    }
}
 

import org.apache.synapse.MessageContext; //导入依赖的package包/类
public void connect(MessageContext messageContext) {
    boolean includeParentDirectory;
    String source = (String) ConnectorUtils.lookupTemplateParamater(messageContext,
            FileConstants.FILE_LOCATION);
    String destination = (String) ConnectorUtils.lookupTemplateParamater(messageContext,
            FileConstants.NEW_FILE_LOCATION);
    String filePattern = (String) ConnectorUtils.lookupTemplateParamater(messageContext,
            FileConstants.FILE_PATTERN);
    String includeParentDir = (String) ConnectorUtils.lookupTemplateParamater(messageContext,
            FileConstants.INCLUDE_PARENT_DIRECTORY);
    if (StringUtils.isEmpty(includeParentDir)) {
        includeParentDirectory = Boolean.parseBoolean(FileConstants.DEFAULT_INCLUDE_PARENT_DIRECTORY);
    } else {
        includeParentDirectory = Boolean.parseBoolean(includeParentDir);
    }
    boolean resultStatus = moveFile(source, destination, messageContext, includeParentDirectory, filePattern);
    generateResults(messageContext, resultStatus);
}
 

import org.apache.synapse.MessageContext; //导入依赖的package包/类
/**
 * @param fileObj        source fileObject
 * @param directoryToZip destination fileObject
 * @param fileList       list of files to be compressed
 * @param messageContext The message context that is generated for processing the file
 * @throws IOException
 */
private void writeZipFiles(FileObject fileObj, FileObject directoryToZip, List<FileObject> fileList,
                           MessageContext messageContext) throws IOException {
    ZipOutputStream zos = null;
    try {
        zos = new ZipOutputStream(directoryToZip.getContent().getOutputStream());
        for (FileObject file : fileList) {
            if (file.getType() == FileType.FILE) {
                addToZip(fileObj, file, zos);
            }
        }
    } catch (IOException e) {
        handleException("Error occur in writing files", e, messageContext);
    } finally {
        if (zos != null) {
            zos.close();
        }
    }
}
 

import org.apache.synapse.MessageContext; //导入依赖的package包/类
public void connect(MessageContext messageContext) {
    String proxyHost = (String) ConnectorUtils
            .lookupTemplateParamater(messageContext, FileConstants.PROXY_HOST);
    String proxyPort = (String) ConnectorUtils
            .lookupTemplateParamater(messageContext, FileConstants.PROXY_PORT);
    String proxyUsername = (String) ConnectorUtils
            .lookupTemplateParamater(messageContext, FileConstants.PROXY_USERNAME);
    String proxyPassword = (String) ConnectorUtils
            .lookupTemplateParamater(messageContext, FileConstants.PROXY_PASSWORD);
    String ftpServer = (String) ConnectorUtils
            .lookupTemplateParamater(messageContext, FileConstants.FTP_SERVER);
    String ftpPort = (String) ConnectorUtils
            .lookupTemplateParamater(messageContext, FileConstants.FTP_PORT);
    String ftpUsername = (String) ConnectorUtils
            .lookupTemplateParamater(messageContext, FileConstants.FTP_USERNAME);
    String ftpPassword = (String) ConnectorUtils
            .lookupTemplateParamater(messageContext, FileConstants.FTP_PASSWORD);

    boolean resultStatus =
            ftpOverHttp(proxyHost, proxyPort, proxyUsername, proxyPassword, ftpServer, ftpPort,
                    ftpUsername, ftpPassword, messageContext);
    generateResult(messageContext, resultStatus);
}
 

import org.apache.synapse.MessageContext; //导入依赖的package包/类
public void connect(MessageContext messageContext) {
    String fileLocation = (String) ConnectorUtils.lookupTemplateParamater(messageContext,
            FileConstants.FILE_LOCATION);
    String chunkSize = (String) ConnectorUtils.lookupTemplateParamater(messageContext,
            FileConstants.CHUNK_SIZE);
    String destination = (String) ConnectorUtils.lookupTemplateParamater(messageContext,
            FileConstants.NEW_FILE_LOCATION);
    String numberOfLines = (String) ConnectorUtils.lookupTemplateParamater(messageContext,
            FileConstants.NUMBER_OF_LINES);
    String xpathExpression = (String) ConnectorUtils.lookupTemplateParamater(messageContext,
            FileConstants.XPATH_EXPRESSION);
    FileSystemOptions options = FileConnectorUtils.init(messageContext);
    boolean resultStatus = splitFile(fileLocation, chunkSize, destination, numberOfLines, xpathExpression, options,
            messageContext);
    generateOutput(messageContext, resultStatus);
}
 

import org.apache.synapse.MessageContext; //导入依赖的package包/类
/**
 * Throw Synapse Exception for any exception in class mediator
 * so that the fault handler will be invoked
 *
 * @param ERROR_CODE
 * @param ERROR_MESSAGE
 * @param ERROR_DETAIL
 * @param context
 */
public static void handle(String ERROR_CODE, String ERROR_MESSAGE, String ERROR_DETAIL, MessageContext context) {

    int array[] = {20, 20, 40};
    int total = 0;
    try {
        for (int i = 5; i >= 0; i--) {
            total += array[i];
        }
    } catch (Exception e) {
        context.setProperty(ERROR_CODE, "AB005");
        context.setProperty(ERROR_MESSAGE, "Error Message from class CsvValidatorMediator");
        context.setProperty(ERROR_DETAIL, "Error Details from class");

        String messageContextErrorCode = (String) context.getProperty(ERROR_CODE);
        String messageContextErrorMessage = (String) context.getProperty(ERROR_MESSAGE);
        String messageContextErrorDetail = (String) context.getProperty(ERROR_DETAIL);
        String separator = "?";

        String concatenatedMessage = (messageContextErrorCode + separator + messageContextErrorMessage + separator + messageContextErrorDetail);
        throw new SynapseException(concatenatedMessage);
    }
}
 

import org.apache.synapse.MessageContext; //导入依赖的package包/类
public boolean mediate(MessageContext msgCtx) {
    try {
        log.debug("BinaryExtractMediator Process, with offset: "+offset+" ,length "+length);
        SOAPBody soapBody = msgCtx.getEnvelope().getBody();
        OMElement firstElement = soapBody.getFirstElement();
        log.debug("First Element : "+firstElement.getLocalName());
        log.debug("First Element Text : "+firstElement.getText());
        OMText binaryNode =(OMText) firstElement.getFirstOMChild();
        log.debug("First Element Node Text : "+binaryNode.getText());
        DataHandler dataHandler =(DataHandler) binaryNode.getDataHandler();
        InputStream inputStream = dataHandler.getInputStream();
        byte[] searchByte = new byte[length];
        inputStream.skip(offset - 1);
        int readBytes = inputStream.read(searchByte,0,length);
        String outString = new String(searchByte,binaryEncoding);
        msgCtx.setProperty(variableName,outString);
        log.debug("Set property to MsgCtx, "+variableName+" = "+outString);
        inputStream.close();
    } catch (IOException e) {
        log.error("Excepton on mediation : "+e.getMessage());
    }
    return true;
}
 

import org.apache.synapse.MessageContext; //导入依赖的package包/类
/**
 * In this method we are dispatching the message to tomcat transport.
 *
 * @param endpoint       Endpoint
 * @param uri            uri
 * @param messageContext message context
 * @return boolean
 */
private boolean dispatchMessage(String endpoint, String uri, MessageContext messageContext) {
    // Adding preserver Headers
    if (passThroughSenderManager != null && passThroughSenderManager.getSharedPassThroughHttpSender() != null) {
        try {
            passThroughSenderManager.getSharedPassThroughHttpSender().addPreserveHttpHeader(HTTP.USER_AGENT);
            // This catch is added when there is no preserve headers in the PassthoughHttpSender.
        } catch (ArrayIndexOutOfBoundsException e) {
            if (log.isDebugEnabled()) {
                log.debug("ArrayIndexOutOfBoundsException exception occurred, when adding preserve headers.");
            }
        }
    }
    if (log.isDebugEnabled()) {
        log.debug("Dispatching message to " + uri);
    }
    messageContext.setProperty(MESSAGE_DISPATCHED, "true");
    Utils.setIntegratorHeader(messageContext, uri);
    setREST_URL_POSTFIX(((Axis2MessageContext) messageContext).getAxis2MessageContext(), uri);
    sendMediator.setEndpoint(Utils.createEndpoint(endpoint, messageContext.getEnvironment()));
    return sendMediator.mediate(messageContext);
}
 

import org.apache.synapse.MessageContext; //导入依赖的package包/类
@Test(description = "Handle request with device type URI with Mutual Auth Header",
        dependsOnMethods = "testHandleSuccessRequestProxyMutualAuthHeader")
public void testHandleSuccessRequestMutualAuthHeader() throws Exception {
    HashMap<String, String> transportHeaders = new HashMap<>();
    transportHeaders.put(AuthConstants.MUTUAL_AUTH_HEADER, "Test Header");
    setMockClient();
    this.mockClient.setResponse(getAccessTokenReponse());
    this.mockClient.setResponse(getValidationResponse());
    MessageContext messageContext = createSynapseMessageContext("<empty/>", this.synapseConfiguration,
            transportHeaders, "https://test.com/testservice/api/testdevice");
    org.apache.axis2.context.MessageContext axisMC = ((Axis2MessageContext) messageContext).getAxis2MessageContext();
    String certStr = getContent(TestUtils.getAbsolutePathOfConfig("ra_cert.pem"));
    X509Certificate cert = X509Certificate.getInstance(new ByteArrayInputStream(certStr.
            getBytes(StandardCharsets.UTF_8.name())));
    axisMC.setProperty(AuthConstants.CLIENT_CERTIFICATE, new X509Certificate[]{cert});
    boolean response = this.handler.handleRequest(messageContext);
    Assert.assertTrue(response);
    this.mockClient.reset();
}
 

import org.apache.synapse.MessageContext; //导入依赖的package包/类
@Override
public boolean mediate(MessageContext messageContext) {

    SynapseLog synLog = getLog(messageContext);

    if (synLog.isTraceOrDebugEnabled()) {
        synLog.traceOrDebug("Start : predict mediator");

        if (synLog.isTraceTraceEnabled()) {
            synLog.traceTrace("Message : " + messageContext.getEnvelope());
        }
    }

    String prediction = getPredictionFromModel(messageContext);
    messageContext.setProperty(resultPropertyName, prediction);

    synLog.traceOrDebug("End : predict mediator");
    this.isUpdated = false;
    return true;
}
 

import org.apache.synapse.MessageContext; //导入依赖的package包/类
/**
 * Get the predicted value for the given input features using the ML-Model
 * @param messageContext    the incoming message context
 * @return                  the predicted value as String
 */
public String getPrediction(MessageContext messageContext) throws MLModelBuilderException, JaxenException,
        MLModelHandlerException {

    String data[] = new String[featureIndexMap.size()];
    for(Map.Entry<SynapsePath, Integer> entry : featureIndexMap.entrySet()) {

        SynapsePath synapsePath = entry.getKey();
        // Extract the feature value from the message
        String variableValue = synapsePath.stringValueOf(messageContext);
        if(variableValue != null) {
            // Get the mapping feature index of the ML-model
            int featureIndex = entry.getValue();
            data[featureIndex] = variableValue;
        }
    }
    return predict(data);
}
 

import org.apache.synapse.MessageContext; //导入依赖的package包/类
private void prepareToRetry(MessageContext messageContext) {
    if (!isTerminated) {
        // First stop the processor since no point in re-triggering jobs if the we can't send
        // it to the client
        if (!messageProcessor.isPaused()) {
            this.messageProcessor.pauseService();

            log.info("Pausing the service of message processor [" + messageProcessor.getName() + "]");
        }

        checkAndDeactivateProcessor(attemptCount, maxDeliverAttempts, messageContext);

        if (log.isDebugEnabled()) {
            log.debug("Failed to send to client retrying after " + retryInterval +
                    "s with attempt count - " + attemptCount);
        }

        try {
            // wait for some time before retrying
            Thread.sleep(retryInterval);
        } catch (InterruptedException ignore) {
            // No harm even it gets interrupted. So nothing to handle.
        }
    }
}
 

import org.apache.synapse.MessageContext; //导入依赖的package包/类
private void checkAndDeactivateProcessor(int attemptCount, int maxAttempts, MessageContext messageContext) {
    if (maxAttempts > 0) {
        this.attemptCount++;

        if (attemptCount >= maxAttempts) {

            if (this.isMaxDeliveryAttemptDropEnabled) {
                continueMessageProcessor();
                if (log.isDebugEnabled()) {
                    log.debug("Message processor ["
                            + messageProcessor.getName()
                            + "] Dropped the failed message and continue due to reach of max attempts");
                }
            }  else  {
                    sendThroughReplySeq(messageContext);
                    continueMessageProcessor();
                if (log.isDebugEnabled()) {
                    log.debug("Message processor ["
                            + messageProcessor.getName()
                            + "] stopped due to reach of max attempts");
                }
                }
            }
        }
}
 

import org.apache.synapse.MessageContext; //导入依赖的package包/类
public boolean mediate(MessageContext messageContext) {
    try {
        if (log.isDebugEnabled()) {
            log.debug("Response interceptor mediation started");
        }
        String clusterId = (String) messageContext.getProperty(LoadBalancerConstants.CLUSTER_ID);
        if (StringUtils.isNotBlank(clusterId)) {
            FutureTask<Object> task = new FutureTask<Object>(new InFlightRequestDecrementCallable(clusterId));
            LoadBalancerStatisticsExecutor.getInstance().getService().submit(task);
        } else {
            if (log.isDebugEnabled()) {
                log.debug("Could not decrement in-flight request count : cluster id not found in message context");
            }
        }

    } catch (Exception e) {
        if (log.isErrorEnabled()) {
            log.error("Could not decrement in-flight request count", e);
        }
    }
    return true;
}
 

import org.apache.synapse.MessageContext; //导入依赖的package包/类
/**
 * Setup load balancer message context properties to be used by the out block of the main sequence.
 * These values will be used to update the Location value in the response header.
 *
 * @param synCtx
 * @param currentMember
 */
private void setupLoadBalancerContextProperties(MessageContext synCtx, org.apache.axis2.clustering.Member currentMember) {
    String targetHostname = extractTargetHost(synCtx);
    org.apache.axis2.context.MessageContext axis2MsgCtx = ((Axis2MessageContext) synCtx).getAxis2MessageContext();

    String httpTransportName = "http", httpsTransportName = "https";
    String transportId = getTransportId(extractIncomingTransport(synCtx));

    if (transportId != null) {
        httpsTransportName = httpsTransportName.concat(transportId);
        httpTransportName = httpTransportName.concat(transportId);
    }

    TransportInDescription httpTransportIn = axis2MsgCtx.getConfigurationContext().getAxisConfiguration().getTransportIn(httpTransportName);
    TransportInDescription httpsTransportIn = axis2MsgCtx.getConfigurationContext().getAxisConfiguration().getTransportIn(httpsTransportName);
    String lbHttpPort = (String) httpTransportIn.getParameter("port").getValue();
    String lbHttpsPort = (String) httpsTransportIn.getParameter("port").getValue();
    String clusterId = currentMember.getProperties().getProperty(LoadBalancerConstants.CLUSTER_ID);

    synCtx.setProperty(LoadBalancerConstants.LB_TARGET_HOSTNAME, targetHostname);
    synCtx.setProperty(LoadBalancerConstants.LB_HTTP_PORT, lbHttpPort);
    synCtx.setProperty(LoadBalancerConstants.LB_HTTPS_PORT, lbHttpsPort);
    synCtx.setProperty(LoadBalancerConstants.CLUSTER_ID, clusterId);
}
 

import org.apache.synapse.MessageContext; //导入依赖的package包/类
/**
 * Update http/https port in axis2 member according to incoming request port.
 *
 * @param synCtx
 * @param axis2Member
 */
private void updateAxis2MemberPorts(MessageContext synCtx, org.apache.axis2.clustering.Member axis2Member) {
    if (log.isDebugEnabled()) {
        log.debug("Updating axis2 member port");
    }

    // Find mapping outgoing port for incoming port
    int incomingPort = findIncomingPort(synCtx);
    String transport = extractTransport(synCtx);
    Port outgoingPort = findOutgoingPort(synCtx, axis2Member, transport, incomingPort);
    if (outgoingPort == null) {
        if (log.isErrorEnabled()) {
            log.error(String.format("Could not find the port for proxy port %d in member %s", incomingPort,
                    axis2Member.getProperties().getProperty(LoadBalancerConstants.MEMBER_ID)));
        }
        throwSynapseException(synCtx, 500, "Internal server error");
    }
    if (LoadBalancerConstants.HTTP.equals(transport)) {
        axis2Member.setHttpPort(outgoingPort.getValue());
    } else if (LoadBalancerConstants.HTTPS.equals(transport)) {
        axis2Member.setHttpsPort(outgoingPort.getValue());
    }
}
 

import org.apache.synapse.MessageContext; //导入依赖的package包/类
/**
 * Find incoming port from request URL.
 *
 * @param synCtx
 * @return
 * @throws MalformedURLException
 */
private int findIncomingPort(MessageContext synCtx) {
    org.apache.axis2.context.MessageContext msgCtx =
            ((Axis2MessageContext) synCtx).getAxis2MessageContext();
    try {
        String servicePrefix = (String) msgCtx.getProperty(LoadBalancerConstants.AXIS2_MSG_CTX_SERVICE_PREFIX);
        if (servicePrefix == null) {
            if (log.isErrorEnabled()) {
                log.error(String.format("%s property not found in axis2 message context", LoadBalancerConstants.AXIS2_MSG_CTX_SERVICE_PREFIX));
            }
            throwSynapseException(synCtx, 500, "Internal server error");
        }
        URL servicePrefixUrl = new URL(servicePrefix);
        return servicePrefixUrl.getPort();
    } catch (MalformedURLException e) {
        if (log.isErrorEnabled()) {
            log.error("Could not find incoming request port");
        }
        throwSynapseException(synCtx, 500, "Internal server error");
    }
    return -1;
}