Common method to prepare the request params for batch operation for both sync and async calls

@param batchOperation
the batch operation
@return IntuitMessage the intuit message
@throws FMSException
342446
private <T extends IEntity> IntuitMessage prepareBatch(BatchOperation batchOperation) throws FMSException {
        IntuitMessage intuitMessage = new IntuitMessage();
        RequestElements requestElements = intuitMessage.getRequestElements();

        //set the request params
        Map<String, String> requestParameters = requestElements.getRequestParameters();
        requestParameters.put(RequestElements.REQ_PARAM_METHOD_TYPE, MethodType.POST.toString());

        IntuitBatchRequest intuitBatchRequest = new IntuitBatchRequest();
        intuitBatchRequest.setBatchItemRequest(batchOperation.getBatchItemRequests());
        
        requestElements.setAction(OperationType.BATCH.toString());
        requestElements.setContext(context);
        requestElements.setObjectToSerialize(getSerializableRequestObject(intuitBatchRequest));
        requestElements.setBatchOperation(batchOperation);

        return intuitMessage;
    }
948158
function IceCandidateBatchingSignallingService(asyncExecService, signallingService, batchingDelayMs) {

    Utils.checkArguments(arguments, 3);

    var queuedCandidates = {};
    var deliveryPromises = {};

    this.on = function () {
        return signallingService.on.apply(signallingService, arguments);
    };

    this.removeAllListeners = function () {
        return signallingService.removeAllListeners.apply(signallingService, arguments);
    };

    this.connect = function () {
        return signallingService.connect.apply(signallingService, arguments);
    };

    this.sendOffer = function () {
        return signallingService.sendOffer.apply(signallingService, arguments);
    };

    this.waitForAnswer = function (correlationId) {
        return signallingService.waitForAnswer.apply(signallingService, arguments);
    };

    this.createNewPointer = function () {
        return signallingService.createNewPointer.apply(signallingService, arguments);
    };

    this.getLocalId = function () {
        return signallingService.getLocalId.apply(signallingService, arguments);
    };

    this.sendAnswer = function () {
        return signallingService.sendAnswer.apply(signallingService, arguments);
    };

    this.sendIceCandidates = function (destinationNode, correlationId, iceCandidates) {
        var newQueue = iceCandidates;
        var existingQueue = getQueue(destinationNode.id, correlationId);
        if (existingQueue) {
            newQueue = existingQueue.concat(iceCandidates);
        }
        else {
            // This is the first set of candidates to be queued for the destination/correlationId combo, schedule their delivery
            setPromise(destinationNode.id, correlationId, scheduleCandidateDelivery(destinationNode, correlationId));
        }
        setQueue(destinationNode.id, correlationId, newQueue);
        return getPromise(destinationNode.id, correlationId);
    };

    function getPromise(destinationNodeId, correlationId) {
        if (deliveryPromises.hasOwnProperty(destinationNodeId) && deliveryPromises[destinationNodeId].hasOwnProperty(correlationId)) {
            return deliveryPromises[destinationNodeId][correlationId];
        }
        else {
            throw new Error("Couldn't locate promise for send?! (this should never happen)");
        }
    }

    function setPromise(nodeId, correlationId, newPromise) {
        if (!deliveryPromises.hasOwnProperty(nodeId)) {
            deliveryPromises[nodeId] = {};
        }
        deliveryPromises[nodeId][correlationId] = newPromise;
    }

    function deletePromise(nodeId, correlationId) {
        delete deliveryPromises[nodeId][correlationId];
        if (Object.keys(deliveryPromises[nodeId]).length === 0) {
            delete deliveryPromises[nodeId];
        }
    }

    function getQueue(nodeId, correlationId) {
        if (queuedCandidates.hasOwnProperty(nodeId) && queuedCandidates[nodeId].hasOwnProperty(correlationId)) {
            return queuedCandidates[nodeId][correlationId];
        }
        return null;
    }

    function setQueue(nodeId, correlationId, newQueue) {
        if (!queuedCandidates.hasOwnProperty(nodeId)) {
            queuedCandidates[nodeId] = {};
        }
        queuedCandidates[nodeId][correlationId] = newQueue;
    }

    function deleteQueue(nodeId, correlationId) {
        delete queuedCandidates[nodeId][correlationId];
        if (Object.keys(queuedCandidates[nodeId]).length === 0) {
            delete queuedCandidates[nodeId];
        }
    }

    function scheduleCandidateDelivery(destinationNode, correlationId) {
        return new Promise(function (resolve, reject) {
            asyncExecService.setTimeout(function () {
                var queueToSend = getQueue(destinationNode.id, correlationId);
                deleteQueue(destinationNode.id, correlationId);
                deletePromise(destinationNode.id, correlationId);
                signallingService.sendIceCandidates(destinationNode, correlationId, queueToSend)
                    .then(resolve, reject);
            }, batchingDelayMs);
        });
    }
}
914923
public static ModelNode createBatchRequest(ModelNode... steps) {
        final ModelNode composite = new ModelNode();
        composite.get(OPERATION).set(BATCH);
        composite.get(ADDRESS).setEmptyList();
        final ModelNode stepsNode = composite.get(BATCH_STEPS);
        for (ModelNode step : steps) {
            if (step != null) {
                stepsNode.add(step);
            }
        }
        return composite;
    }
418666
public JSONObject batch(List<JSONObject> actions) throws AlgoliaException {
    return postBatch(actions, RequestOptions.empty);
  }
389105
public final void persist(BatchingContext batchingContext, TransactionState tranState)
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.entry(this, tc, "persist", new Object[] { batchingContext, tranState});

        if ((tranState == TransactionState.STATE_COMMITTED)
            || (tranState == TransactionState.STATE_COMMITTING_1PC)
            || (tranState == TransactionState.STATE_COMMITTING_2PC))
        {
            batchingContext.delete(getPersistable());
        }
        else if ((tranState == TransactionState.STATE_PREPARED)
                 || (tranState == TransactionState.STATE_PREPARING))
        {
            Persistable tuple = getPersistable();
            tuple.setLogicallyDeleted(true);
            batchingContext.updateLogicalDeleteAndXID(tuple);
        }
        else if ((tranState == TransactionState.STATE_ROLLEDBACK)
                 || (tranState == TransactionState.STATE_ROLLINGBACK))
        {
            Persistable tuple = getPersistable();
            tuple.setLogicallyDeleted(false);
            batchingContext.updateLogicalDeleteAndXID(tuple);
        }
        else
        {
            if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.exit(this, tc, "persist");
            throw new IllegalStateException(nls.getFormattedMessage("INVALID_TASK_OPERATION_SIMS1520", new Object[] {tranState}, null));
        }

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.exit(this, tc, "persist");
    }
869328
@Override
	public void executeBatch(OperationsQueue queue) {
		if ( batchableGridDialect != null ) {
			batchableGridDialect.executeBatch( queue );
		}
		else if ( groupingByEntityGridDialect != null ) {
			groupingByEntityGridDialect.executeBatch( queue );
		}
	}
971045
public void executeQueryAsync(String query, CallbackHandler callbackHandler) throws FMSException {
        IntuitMessage intuitMessage = prepareQuery(query);

        //set callback handler
        intuitMessage.getRequestElements().setCallbackHandler(callbackHandler);

        //execute async interceptors
        executeAsyncInterceptors(intuitMessage);
    }
347285
public static Result newBatchedExecuteRequest() {

        Type[] types  = new Type[]{ Type.SQL_VARCHAR };
        Result result = newResult(ResultConstants.BATCHEXECDIRECT);

        result.metaData = ResultMetaData.newSimpleResultMetaData(types);

        return result;
    }
567587
@SuppressWarnings("unchecked")
    public <T extends IEntity> T voidRequest(T entity) throws FMSException {

        IntuitMessage intuitMessage = prepareVoidRequest(entity);

        //execute interceptors
        executeInterceptors(intuitMessage);

        return (T) retrieveEntity(intuitMessage);
    }
1320409
public ResponseFromManager generateUpdateExecuteTask(ParallelTask task) {

        // add to map now; as can only pass final
        ParallelTaskManager.getInstance().addTaskToInProgressMap(
                task.getTaskId(), task);
        logger.info("Added task {} to the running inprogress map...",
                task.getTaskId());

        boolean useReplacementVarMap = false;
        boolean useReplacementVarMapNodeSpecific = false;
        Map<String, StrStrMap> replacementVarMapNodeSpecific = null;
        Map<String, String> replacementVarMap = null;

        ResponseFromManager batchResponseFromManager = null;

        switch (task.getRequestReplacementType()) {
        case UNIFORM_VAR_REPLACEMENT:
            useReplacementVarMap = true;
            useReplacementVarMapNodeSpecific = false;
            replacementVarMap = task.getReplacementVarMap();
            break;
        case TARGET_HOST_SPECIFIC_VAR_REPLACEMENT:
            useReplacementVarMap = false;
            useReplacementVarMapNodeSpecific = true;
            replacementVarMapNodeSpecific = task
                    .getReplacementVarMapNodeSpecific();
            break;
        case NO_REPLACEMENT:
            useReplacementVarMap = false;
            useReplacementVarMapNodeSpecific = false;
            break;
        default:
            logger.error("error request replacement type. default as no replacement");
        }// end switch

        // generate content in nodedata
        InternalDataProvider dp = InternalDataProvider.getInstance();
        dp.genNodeDataMap(task);

        VarReplacementProvider.getInstance()
                .updateRequestWithReplacement(task, useReplacementVarMap,
                        replacementVarMap, useReplacementVarMapNodeSpecific,
                        replacementVarMapNodeSpecific);

        batchResponseFromManager = 
                sendTaskToExecutionManager(task);

        removeTaskFromInProgressMap(task.getTaskId());
        logger.info(
                "Removed task {} from the running inprogress map... "
                        + ". This task should be garbage collected if there are no other pointers.",
                task.getTaskId());
        return batchResponseFromManager;

    }
176395
private Future<?> sendOperation(List<Data> keyValueSequence) {
        OperationService operationService = mapServiceContext.getNodeEngine().getOperationService();
        Operation operation = createOperation(keyValueSequence);
        return operationService.invokeOnPartition(MapService.SERVICE_NAME, operation, partitionId);
    }
18566
public <T extends IEntity> void uploadAsync(T entity, InputStream docContent, CallbackHandler callbackHandler) throws FMSException {

        IntuitMessage intuitMessage = prepareUpload(entity, docContent);

        //set callback handler
        intuitMessage.getRequestElements().setCallbackHandler(callbackHandler);

        //execute async interceptors
        executeAsyncInterceptors(intuitMessage);
    }
292543
public void execute(List<IntuitMessage> intuitMessages) throws FMSException {
        LOG.debug("Enter HTTPBatchClientConnectionInterceptor - batch...");
        
        RequestElements intuitRequest = getFirst(intuitMessages).getRequestElements();
        
        IntuitRetryPolicyHandler handler = getRetryHandler();
        HttpClientBuilder hcBuilder = HttpClients.custom()
                .setRetryHandler(handler)
                .setDefaultRequestConfig(setTimeout(intuitRequest.getContext()))
                .setDefaultCredentialsProvider(setProxyAuthentication())
                .setSSLSocketFactory(prepareClientSSL());
        
        entitiesManager.reset();
        HttpHost proxy = getProxy();

        if (proxy != null) {
            hcBuilder.setProxy(proxy);
        }

        CloseableHttpClient client = hcBuilder.build();
        HttpRequestBase httpRequest = prepareHttpRequest(intuitRequest);

        //Additional processing for POST requests
        if(httpRequest instanceof HttpPost) {
            for (IntuitMessage intuitMessage : intuitMessages) {
                execute(intuitMessage);
            }
            try {
                ((HttpPost) httpRequest).setEntity(entitiesManager.asSingleEntity());
            } catch (IOException ex) {
                LOG.debug("HTTPBatchClientConnectionInterceptor was unable to convert input into single HTTP entity");
                throw new FMSException("Unable to prepare http entity" + ex.getMessage() + "\n" + ex.getStackTrace());
            } finally {
                entitiesManager.reset();
            }


        }

        //Method to send request
        IntuitMessage intuitMessage = executeHttpRequest(httpRequest,client);

        //TODO review this implementation, since it is using one temporary instance to fill up response elements in original
        for (IntuitMessage originalMessage : intuitMessages) {
            ResponseElements originalResponseElements = originalMessage.getResponseElements();
            ResponseElements receivedResponseElements = intuitMessage.getResponseElements();

            originalResponseElements.setResponseContent(   receivedResponseElements.getResponseContent()   );
            originalResponseElements.setEncodingHeader(    receivedResponseElements.getEncodingHeader()    );
            originalResponseElements.setContentTypeHeader( receivedResponseElements.getContentTypeHeader() );
            originalResponseElements.setStatusLine(        receivedResponseElements.getStatusLine()        );
            originalResponseElements.setStatusCode(        receivedResponseElements.getStatusCode()        );

        }
        LOG.debug("Exit HTTPBatchClientConnectionInterceptor.");

    }
834034
public int prepare() throws ProtocolException, RollbackException, SeverePersistenceException, TransactionException
    {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.entry(this, tc, "prepare");

        // Check the current transaction state (use state lock to ensure we see the latest value)
        synchronized (_stateLock)
        {
            if (_state != TransactionState.STATE_ACTIVE)
            {
                ProtocolException pe = new ProtocolException("TRAN_PROTOCOL_ERROR_SIMS1001", new Object[]{});
                if (TraceComponent.isAnyTracingEnabled() && tc.isEventEnabled()) SibTr.event(this, tc, "Cannot prepare Transaction. Transaction is complete or completing!", pe);
                if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.exit(this, tc, "prepare");
                throw pe;
            }

            // Transition our prepare sub-state immediately - before we call the callbacks
            _preCommitSubstate = TransactionState.STATE_PREPARING;
        }

        // PK81848
        // From this point on, we must perform a rollback if we fail to complete the prepare,
        // or another thread requests a deferred rollback
        try
        {
            // Call beforeCompletion on MP callback
            for (int i = 0; i < _callbacks.size(); i++)
            {
                TransactionCallback callback = (TransactionCallback) _callbacks.get(i);
                callback.beforeCompletion(this);
            }

            // Do we have work to do?
            if (_workList != null)
            {
                _workList.preCommit(this);

                synchronized (_stateLock)
                {
                    // Need to change the overalll state after we have 
                    // called the callbacks and preCommitted 
                    // the workList as they may trigger
                    // an addWork call to add work to the
                    // transaction.
                    _state = TransactionState.STATE_PREPARING;
                }

                // The transaction is about to be in a state where any rollback must
                // update the database.  We cannot narrow the window any more than this
                _rollbackChanges = true;
                
                // Perform the prepare
                _persistence.prepare(this);

                // State check/change must be performed holding the state lock
                synchronized (_stateLock)
                {
                    // Another thread may have have received a rollback, while we were
                    // performing our prepare - if the database was extremely slow to respond
                    if (_preCommitSubstate == TransactionState.STATE_ROLLBACK_DEFERRED)
                    {
                        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) SibTr.event(this, tc, "Deferred rollback requested during prepare for TranId: "+_ptid);

                        // Throw an exception to trigger rollback processing.
                        throw new DeferredRollbackException();
                    }
                    else
                    {
                        _state = _preCommitSubstate = TransactionState.STATE_PREPARED;
                    }
                }                
            }
            else
            {
                // State change must be performed holding the state lock
                synchronized (_stateLock)
                {
                    // Defect 301480
                    // We don't have any work in our workList so in theory we should be
                    // able to vote READ_ONLY at this point. However in a particular case
                    // (Best Effort message being sent off server) a call back in an empty
                    // XAResource is used to trigger work on a remote server. In this case 
                    // we need beforeCompletion AND afterCompletion to be called at the 
                    // CORRECT time during transaction completion to ensure that the off 
                    // server work stays part of the XA transaction and gets informed of 
                    // the correct completion direction.
                    _state = _preCommitSubstate = TransactionState.STATE_PREPARED;
                }
            }
        }
        catch (SeverePersistenceException spe)
        {
            // SPE exceptions are handled with an RMFAIL return code by the caller,
            // so unlike all other types of exception no attempt is made here to
            // rollback the transaction. However, we report a local error - so
            // the whole JVM will be coming down with a few minutes.
            com.ibm.ws.ffdc.FFDCFilter.processException(spe, "com.ibm.ws.sib.msgstore.transactions.XidParticipant.prepare", "1:420:1.73", this);

            // As we have thrown a severe exception we need to inform
            // the ME that this instance is no longer able to continue.
            if (_ms != null)
            {
                _ms.reportLocalError();
            }
            if (TraceComponent.isAnyTracingEnabled() && tc.isEventEnabled()) SibTr.event(this, tc, "Unrecoverable persistence exception caught during prepare phase of transaction!", spe);
            if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.exit(this, tc, "prepare", spe);
            throw spe;
        }
        // Catch and handle all exceptions but allow 
        // Errors to filter up without interference
        catch (Exception e)
        {
            com.ibm.ws.ffdc.FFDCFilter.processException(e, "com.ibm.ws.sib.msgstore.transactions.XidParticipant.prepare", "1:436:1.73", this);
            if (TraceComponent.isAnyTracingEnabled() && tc.isEventEnabled()) SibTr.event(this, tc, "Transaction will roll-back due to Exception in prepare!", e);

            // Update our prepare state before calling rollback. If we are responding
            // to a deferred rollback request this will have no affect but if this is 
            // due to some other failure it will lock out any other attempts to rollback
            synchronized (_stateLock)
            {
                _preCommitSubstate = TransactionState.STATE_ROLLINGBACK;
            }

            // An error has occurred but we are able to keep the ME
            // going so we just need to clean up this transaction.
            try
            {
                // When we throw our RollbackException the TM is
                // going to assume we are complete so we need to 
                // ensure we are rolled-back.
                // 
                // PK81848.2
                // We need to pass "true" in order to turn off the
                // state checking which should block other threads 
                // from attempting a rollback when we are in the 
                // middle of internal rollback processing.
                rollback(true);
            }
            // Catch and handle all exceptions but allow 
            // Errors to filter up without interference
            catch (Exception e2)
            {
                com.ibm.ws.ffdc.FFDCFilter.processException(e2, "com.ibm.ws.sib.msgstore.transactions.XIDParticipant.prepare", "1:466:1.73", this);
                
                // This is a problem because we now must RMFAIL this prepare call as we do not know the outcome.
                // Transition to rollback expected state now even if we are in deferred rollback.  This is the
                // only thing we can do (problem is we said XA_OK to roll back in deferred case and now we
                // cannot honour it).  This case may still leave indoubts.
                synchronized(_stateLock)
                {
                    _state = _preCommitSubstate = TransactionState.STATE_ROLLBACK_EXPECTED;
                }
                
                if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.exit(this, tc, "prepare", "Exception caught rolling-back transaction after preCommit failure! " + e2);
                throw new TransactionException(e2);
            }

            // Mark ourselves as complete, regardless of the outcome of the attempt to rollback.
            synchronized (_stateLock)
            {
                _state = _preCommitSubstate = TransactionState.STATE_ROLLEDBACK;
            }

            // Wrap any persistence exception, or an unexpected exception, in a rollback exception.
            // This informs the TM that they can forget about this transaction.
            RollbackException re = new RollbackException("COMPLETION_EXCEPTION_SIMS1002", new Object[]{e}, e);
            if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.exit(this, tc, "prepare", re);
            throw re;
        }

        // Non-exception return path always returns XA_OK
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.exit(this, tc, "prepare", "return="+XAResource.XA_OK);
        return XAResource.XA_OK;
    }
1122794
@Override
	public Void call() throws FMSException {
		CallbackMessage callbackMessage = new CallbackMessage();
		try {
			// add configuration properties from the caller thread
			Config.addConfigurationOverrides(configuration);
			executeInterceptors(intuitMessage);
		} catch (FMSException e) {
			callbackMessage.setFMSException(e);
			LOG.error("Exception in interceptor flow", e);
		}
		// set callbackMessage
		intuitMessage.getResponseElements().setCallbackMessage(callbackMessage);
		
		// execute callback handler interceptor
		new CallbackHandlerInterceptor().execute(intuitMessage);
		
		return null;
	}
736020
public static TransactionalProtocolClient.Operation wrap(final ModelNode operation, final OperationMessageHandler messageHandler, final OperationAttachments attachments) {
        return new TransactionalOperationImpl(operation, messageHandler, attachments);
    }
1283118
private String buildRequestParams(Map<String, String> requestParameters) throws FMSException {
		StringBuilder reqParams = new StringBuilder();
		Set<String> keySet = requestParameters.keySet();
		Iterator<String> keySetIterator = keySet.iterator();

		while (keySetIterator.hasNext()) {
			String key = keySetIterator.next();
			String value = requestParameters.get(key);
			if (isKeyValueExpected(key)) {
//Changes for France and Minor Version 5
				if (value=="updateAccountOnTxns"){
					reqParams.append("include").append("=").append("updateaccountontxns").append("&");
					}
				else if (value=="donotUpdateAccountOnTxns"){
					reqParams.append("include").append("=").append("donotupdateaccountontxns").append("&");
					}
				else
					{
						reqParams.append(key).append("=").append(value).append("&");
					}
		
			} else if (key.equals(RequestElements.REQ_PARAM_QUERY) || key.equals(RequestElements.REQ_PARAM_ENTITIES)) {
				try {
					String encodedQuery = URLEncoder.encode(value, "UTF-8");
					reqParams.append(key).append("=").append(encodedQuery).append("&");
				} catch (UnsupportedEncodingException e) {
					throw new FMSException("UnsupportedEncodingException", e);
				}
			}
		}

		return reqParams.toString();
	}