//------------------------------------------------------------------------------
// <copyright file="BrokerClient.java" company="Microsoft">
//      Copyright (c) Microsoft Corporation.  All rights reserved.
// </copyright>
// <summary>
//      Implement BrokerClient class.
// </summary>
//------------------------------------------------------------------------------

package com.microsoft.hpc.scheduler.session;

import java.io.*;
import java.lang.reflect.*;
import java.net.URL;
import java.util.*;

import javax.xml.bind.*;
import javax.xml.namespace.QName;
import javax.xml.soap.*;
import javax.xml.transform.*;
import javax.xml.transform.dom.DOMSource;
import javax.xml.transform.stream.StreamResult;
import javax.xml.ws.*;
import javax.xml.ws.soap.AddressingFeature;

import org.apache.cxf.endpoint.Client;
import org.apache.cxf.frontend.ClientProxy;
import org.apache.cxf.jaxws.JaxWsServerFactoryBean;
import org.apache.cxf.transport.http.HTTPConduit;
import org.apache.cxf.transports.http.configuration.HTTPClientPolicy;
import org.apache.cxf.binding.Binding;

import com.microsoft.hpc.*;
import com.microsoft.hpc.sessionlauncher.*;
import com.microsoft.hpc.brokercontroller.*;
import com.microsoft.hpc.exceptions.*;
import com.microsoft.hpc.scheduler.session.internal.*;
import com.microsoft.schemas._2003._10.serialization.arrays.ArrayOfstring;

/**
 * Used to communicate with the broker to send and receive messages
 * 
 * @see #sendRequest(Object)
 * @see #endRequests()
 * @see #pullResponses(Class)
 */
public class BrokerClient<TContract extends Service>
{
     // private object objectLock = new object();
     //     objectLock is unused
     
     // private IOutputChannel brokerClient;
     //     BrokerClient uses dispatches to send requests, rather than speaking to the broker directly
     
     // private OperationDescriptionCollection operations;
     //     Java does not use operations
    
    /**
     * Store the count of the sent messages
     */
    private int sendCount = 0;
    
    // private CallbackManager callbackManager
    //      Callbacks are not supported in Java
    
    // private int defaultResponsesTimeout = 60 * 1000;
    //      PullResponses uses a built in timeout of 1 minute
    
    /**
     * Binding used to communicate with broker
     */
    private Binding binding;
    
    // private IResponseService responseServiceClient;
    // private DuplexChannelFactory<IResponseService> responseServiceChannelFactory;
    //      These fields were required for GetResponses which is not implemented for Java
    
    /**
     * Reference to Controller client.
     * Java Changes: Changed from BrokerControllerClient.
     * Implemented as IBrokerController in Java
     */
    private IBrokerController controllerClient;
    
    /**
     * Reference to the Session object passed to the client
     */
    private SessionBase session;
    
    /**
     * The string of the clientid
     */
    private String clientId;
    
    /**
     * Client Identity QName
     */
    private QName clientIdHeader;
    
    /**
     * Stores the scheme
     */
    private TransportScheme scheme;
    
    //The following fields are only necessary for the java implementation.
    /**
     * Specifies the default timeout for EOM call
     */
    final public int DEFAULT_TIMEOUT = 60000;
    
    /**
     * Specifies the sleep interval between response pulls in PullResponses.
     * <p>Slows down performance, found to not really be necessary, but
     *      if problem does occur, change should be here.</p>
     */
    final public int REQUEST_RESPONSES_INTERVAL = 0;
    
    /**
     * Specifies the initial batch count to attempt
     */
    final public int MAX_INITIAL_BATCH_COUNT = 10000;
    
    /**
     * Stores the type of dispatch to send requests
     */
    private Map<String,Dispatch<SOAPMessage>> dispatchMap = null;
    
    /**
     * Parses the wsdl file to retrieve default action, response, and binding values
     */
    private WSDLParser wsdlFile;
    
    /**
     * Reference for the Service class used to generate the Dispatch
     */
    private TContract service;
    
    /**
     * Creates instances of the BrokerClient
     * @param session Session or DurableSession
     * @param service Service class to instantiate
     */
    public BrokerClient(SessionBase session, Class<TContract> service) 
    {
        Utility.throwIfNull(session, "session");
        
        init(session, null, null, null, service);
    }
    
    /**
     * Creates instances of the BrokerClient
     * @param session Session or DurableSession
     * @param binding Binding to use to connect to service
     * @param service Service class to instantiate
     */
    public BrokerClient(SessionBase session, Binding binding, Class<TContract> service) 
    {
        Utility.throwIfNull(session, "session");
        
        init(session, binding, null, null, service);
    }
    
    /**
     * Creates instances of the BrokerClient
     * @param clientid String identity of the client
     * @param session Session or DurableSession
     * @param service Service class to instantiate
     */
    public BrokerClient(String clientid, SessionBase session, Class<TContract> service)
    {
        Utility.throwIfNull(session, "session");
        Utility.throwIfNullOrEmpty(clientid, "clientid");
        Utility.throwIfTooLong(clientid.length(), "clientid", 128, SR.ClientIdTooLong);
        
        init(session, null, null, clientid, service);
    }
    
    /**
     * Creates instances of the BrokerClient
     * @param clientid String identity of the client
     * @param session Session or DurableSession
     * @param binding Binding to use to connect to the broker
     * @param service Service class to instantiate
     */
    public BrokerClient(String clientid, SessionBase session, Binding binding, Class<TContract> service)
    {
        Utility.throwIfNull(session, "session");
        Utility.throwIfNullOrEmpty(clientid, "clientid");
        Utility.throwIfTooLong(clientid.length(), "clientid", 128, SR.ClientIdTooLong);
        
        init(session, binding, null, clientid, service);
    }
    
    /**
     * Creates instances of the BrokerClient
     * @param session Session or DurableSession
     * @param bindingConfigName Binding configuration to use to connect to broker
     * @param service Service class to instantiate
     */
    public BrokerClient(SessionBase session, String bindingConfigName, Class<TContract> service)
    {
        Utility.throwIfNull(session, "session");
        
        init(session, null, bindingConfigName, null, service);
    }
    
    /**
     * Creates instances of the BrokerClient
     * @param clientid String identity of the client
     * @param session Session or DurableSession
     * @param bindingConfigName Binding configuration to use to connect to the broker
     * @param service Service class to instantiate
     */
    public BrokerClient(String clientid, SessionBase session, String bindingConfigName, Class<TContract> service)
    {
        Utility.throwIfNull(session, "session");
        Utility.throwIfNullOrEmpty(clientid, "clientid");
        Utility.throwIfTooLong(clientid.length(), "clientid", 128, SR.ClientIdTooLong);
        
        init(session, null, bindingConfigName, clientid, service);
    }
    
    /**
     * Initializes BrokerClient session
     * @param session Session or DurableSession
     * @param binding Binding configuration to use to connect to the broker
     * @param bindingConfigName Binding configuration to use to connect to the broker
     * @param _clientId String identity of the client
     * @param service Service class to instantiate
     * @throws Exception
     */
    private void init(SessionBase session, Binding binding, String bindingConfigName, String _clientId, Class<TContract> service)
    {
        //printSessionInfo(session.getInfo());
        if(_clientId == null || _clientId == "")
        {
            clientId = "";
        }
        else
        {
            clientId = _clientId;
        }
        
        //ParamCheckUtility...
        
        this.dispatchMap = new HashMap<String, Dispatch<SOAPMessage>>();
        this.session = session;
        this.binding = binding;
        
        String filename = "";
        try
        {
            Constructor<TContract> serviceCons = service.getConstructor(URL.class);
            filename = "/" + ((URL)service.getDeclaredField("WSDL_LOCATION").get(null)).getPath();
            
            this.service = serviceCons.newInstance(this.getClass().getResource(filename));
        }
        catch(Exception e)
        {
            throw new ArgumentException(SR.RequireServiceContractType);
        }
        
        
        try
        { 
            wsdlFile = new WSDLParser(filename); //try to get specific WSDL location
        }
        catch(Exception e)
        {
            wsdlFile = new WSDLParser(); //if cannot be found, use any WSDL files in current directory
        }
        
        // Try to get the scheme from session info
        this.scheme = TransportScheme.valueOf((session.getInfo().getTransportScheme().get(0)));
            
        // Set the scheme
        if (this.scheme == TransportScheme.None)
        {
            throw new NotSupportedException();
        }
        
        
        // TODO: Check if the scheme is multi, should throw exception (from .NET)
        // Set the binding
        if (bindingConfigName != null && !bindingConfigName.isEmpty())
        {
            this.binding = getConfiguredClientBinding(bindingConfigName);
        }
        else if (this.binding == null)
        {
            this.binding = getDefaultClientBinding(this.service);
        }
        JaxWsServerFactoryBean sf = new JaxWsServerFactoryBean();
        sf.setServiceClass(service);
        sf.getServiceFactory().setWrapped(true);
        sf.setBindingId(this.binding.getBindingInfo().getBindingId()); //unecessary for custom bindings?
        //we never formally create the port on this machine, so may not be necessary.
        
        clientIdHeader = new QName(Constant.HpcHeaderNS, Constant.ClientIdHeaderName);
        
    }
    
    /**
     * Send typed message to broker
     * @param <TMessage> Typed message type
     * @param request Typed message to send
     * @throws SessionException
     * @see {@link HpcJava#createRequest(Class, Object...)} 
     */
    public <TMessage> void sendRequest(TMessage request) throws SessionException 
    {
        sendRequest(request, "");
    }
    
    /**
     * Send typed messages with user data to broker
     * @param <TMessage> Typed message type
     * @param request Typed message to send
     * @param userData User data (will be converted to String because of Java limitations)
     * @throws SessionException 
     * @see {@link HpcJava#createRequest(Class, Object...)}
     */
    public <TMessage> void sendRequest(TMessage request, Object userData) throws SessionException
    {
        Utility.throwIfNull(request, "request");
        Utility.throwIfNull(userData, "userData");
        
        String action = wsdlFile.getAction(request.getClass().getSimpleName());
        Utility.throwIfInvalid(action != null, "TMessage");
        sendRequest(request, userData, action);
    }
    
    /**
     * Send typed messages with user data to broker
     * @param <TMessage> Typed message type
     * @param request Typed message to send
     * @param userData User data (will be converted to String because of Java limitations)
     * @param action Action of typed message if type is ambiguous
     * @throws SessionException 
     * @see {@link HpcJava#createRequest(Class, Object...)}
     */
    public <TMessage> void sendRequest(TMessage request, Object userData, String action) throws SessionException
    {
        Utility.throwIfNull(request, "request");
        Utility.throwIfNull(userData, "userData");
        Utility.throwIfNull(action, "action");
        
        String brokerEpr = generateEndpointAddress(session.getInfo().getBrokerEpr(), this.scheme);
        
        Dispatch<SOAPMessage> dispatch = getSOAPDispatch(this.service,
                //this.bindingName, brokerEpr,
                binding, brokerEpr,
                action);
        
        SOAPMessage soapMessage = createSOAPMessage(request, userData);
        
        dispatch.invokeOneWay(soapMessage);
        sendCount++;
    }
    
    /**
     * Generates SOAP header for user data
     * <p>Java Changes: return type to void, added SOAPMessage msg as parameter
     * Required for interoperability with Java's native SOAP functions</p>
     * @param userData
     * @param msg
     * @throws SOAPException
     */
    private static void generateUserDataHeader(Object userData, SOAPMessage msg) throws SOAPException
    {
        SOAPHeader header = msg.getSOAPHeader();
        SOAPHeaderElement headerElement = header.addHeaderElement(
                    new QName(Constant.HpcHeaderNS, Constant.UserDataHeaderName));
        headerElement.setValue(userData.toString());
    }
    
    // public BrokerResponseEnumerator<TMessage> GetResponses<TMessage>()
    // public BrokerResponseEnumerator<TMessage> GetResponses<TMessage>(int waitTimeoutMilliseconds)
    // public BrokerResponseEnumerator<TMessage> GetResponses<TMessage>(string action, int waitTimeoutMilliseconds)
    //      Not implemented in Java
    
    // public void AddResponseHandler<TMessage>(BrokerResponseHandler<TMessage> callback)
    // public void AddResponseHandler<TMessage>(BrokerResponseHandler<TMessage> callback, string action)
    //      Asynchronous is not supported in Java
    
    //private FaultDescriptionCollection GetFaultDetailsByAction(string action, string replyAction)
    //      faultCollections not used in PullResponses
    
    // private MessageDescription GetMessageDescription(Type messageType, MessageDirection messageDirection)
    //      Not implemented in Java. 
    //      getSOAPMessage(...) may be synonymous with this function
    
    // private MessageDescription GetInputMessageDescriptionFromOutputMessageType(Type messageType, out string replyAction)
    //      No java equivalent
    
    // private IOutputChannel GetBrokerClient()
    //      BrokerClient uses dispatches to send requests, rather than speaking to the broker directly
    
    /**
     * Generate endpoint address by scheme from epr list
     * <p>ava Changes: returns a string rather than a EndpointAddress</p>
     * @param eprList indicating the epr list
     * @param scheme indicating the scheme
     * @return  endpoint address
     */
    private static String generateEndpointAddress(JAXBElement<ArrayOfstring> eprList, TransportScheme scheme)
    {
        int index = scheme.getIndex();
        return eprList.getValue().getString().get(index);
    }
    /**
     * Generate endpoint address by scheme from epr list
     * @param eprList
     * @param scheme
     * @return  endpoint address
     */
    private static String generateEndpointAddress(JAXBElement<ArrayOfstring> eprList, String scheme)
    {
        return generateEndpointAddress(eprList, TransportScheme.valueOf(scheme));
    }
    
    //private IResponseService GetResponseServiceClient()
    //      Not implemented in Java
    
    /**
     * Get client proxy to controller service and caches it
     * @return  IBrokerController
     */
    private IBrokerController getControllerClient()
    {
        if(this.controllerClient == null)
        {
            String brokerControllerAddress = generateEndpointAddress(session.getInfo().getControllerEpr(), this.scheme);
            String wsdl = "/" + BrokerController.WSDL_LOCATION.getPath();
            BrokerController brokerControllerSvc = new BrokerController(this.getClass().getResource(wsdl));
            
            this.controllerClient = brokerControllerSvc.getDefaultBindingIBrokerController();
            
            java.util.Map<String, Object> requestContext = ((javax.xml.ws.BindingProvider)this.controllerClient).getRequestContext();
            requestContext.put(Dispatch.ENDPOINT_ADDRESS_PROPERTY, brokerControllerAddress);
        }
        return this.controllerClient;
    }
    
    /**
     * Returns the default binding to use for service.
     * Java Changes: Uses values defined by CXF code
     * @return  binding
     */
    private Binding getDefaultClientBinding(TContract service)
    {
        Method[] methods = service.getClass().getMethods();
        Object clientService = null;
        for(Method t : methods)
        {
            if(t.getName().startsWith("getDefault") && t.getParameterTypes().length == 0)
            {//searches for a function that starts with "getDefault" (as in getDefaultBindingIEchoSvc)
                try
                {
                    clientService = t.invoke(service);
                    break;
                } catch (Exception e)
                {
                    ;
                }
            }
        }
        if(clientService == null)
        {
            throw new ArgumentException(SR.Broker_BindingNotSupported);
        }
        Client client = ClientProxy.getClient(clientService);
        
        return client.getEndpoint().getBinding();
    }
    
    /**
     * Returns the binding based on the specified configuration name.
     * Java Changes: This function is not supported in Java since Http is the only binding supported
     * @param configurationName
     */
    private Binding getConfiguredClientBinding(String configurationName)
    {
        throw new NotSupportedException("GetConfiguredClientBinding not implemented");
    }
    
    /**
     * Handles cleanup of the BrokerClient instance.
     * Semi-Equivalent to Dispose in C#
     */
    protected void finalize()
    {
        dispose();
    }
    
    /**
     * Handles cleanup of the BrokerClient instance
     */
    protected void dispose()
    {
        this.controllerClient = null;
        this.session = null;
        this.dispatchMap = null;
    }
    
    /**
     * Commits requests to broker's store. 
     * @param timeoutMilliseconds timeout before exception
     * @throws InvalidOperationException
     */
    public void endRequests(int timeoutMilliseconds) 
    throws InvalidOperationException
    {
        if (timeoutMilliseconds < 0)
        {
            throw new ArgumentException(SR.InvalidTimeout, "timeoutMilliseconds");
        }
        if (this.sendCount == 0)
        {
            throw new InvalidOperationException("SendMessage must be called at least once before EndRequests");
        }
        
        controllerClient = getControllerClient();
        
        controllerClient.endRequests(sendCount, timeoutMilliseconds, clientId);
    }
    
    /**
     * Commits requests to broker's store. Default timeout is defined above.
     * @throws InvalidOperationException
     */
    public void endRequests() 
    throws InvalidOperationException
    {
        endRequests(DEFAULT_TIMEOUT);
    }
    
    /**
     * Closes broker connections
     */
    public void close()
    {
        close(false, Constant.PurgeTimeoutMS);
    }
    
    /**
     * Close the connection and purge the results to the server
     * @param purge if true, the result will be removed in the broker
     */
    public void close(Boolean purge)
    {
        close(purge, Constant.PurgeTimeoutMS);
    }
    
    /**
     * Close the connection and purge the results in the server
     * @param purge if true, the result will be removed in the broker
     * @param timeoutMilliseconds the timeout for the response if purge
     */
    public void close(Boolean purge, int timeoutMilliseconds)
    {
        if(purge)
        {
            IBrokerController controller = getControllerClient();
            //setTimeout(controller, timeoutMilliseconds);
            controller.purge(clientId);
        }
        dispose();
    }
    
    //Java implementation functions
    //These functions do not exist in the .NET version
    /**
     * Prints information about a session
     * Useful for debugging
     * @param SessionInfo
     */
    public static void printSessionInfo(SessionInfo s)
    {
        System.out.println("Session Info: ");
        System.out.println(" Id: " + s.getId());
        JAXBElement<ArrayOfstring> epr = s.getBrokerEpr();
        String eprString = generateEndpointAddress(epr, s.getTransportScheme().get(0));
        
        System.out.println(" BrokerEpr: " + eprString);

        JAXBElement<ArrayOfstring> controllerEpr = s.getControllerEpr();
        System.out.println(" controllerEpr: " + generateEndpointAddress(controllerEpr, s.getTransportScheme().get(0)));
        
        JAXBElement<ArrayOfstring> responseEpr = s.getResponseEpr();
        System.out.println(" responseEpr: " + generateEndpointAddress(responseEpr, s.getTransportScheme().get(0)));
        
        String blah = s.getBrokerLauncherEpr().getValue();
        System.out.println(" launcherEpr: " + blah);
        
    }

    /**
     * Retrieves messages from the server (Java ~ GetResponses).
     * 
     * <p>Works by pulling a default maximum of 10,000 messages from the server at a time.
     * If there are more messages on the server, it will continue to pull until all messages have been retrieved
     * or a timeout has occurred.</p>
     *  
     * @param <TMessage> Type of message to retrieve
     * @param responseClass Type of message to retrieve
     * @return a list of BrokerResponse classes
     * @see BrokerResponse
     * @throws TimeoutException if pullResponses is unable to complete before timing out
     * @throws SessionException if there was a session error
     */
    public <TMessage> List<BrokerResponse<TMessage>> pullResponses(Class<TMessage> responseClass) throws TimeoutException, SessionException
    {
        String responseAction = wsdlFile.getResponseAction(responseClass.getSimpleName());
        
        return pullResponses(responseClass, responseAction, sendCount, Timeout.Infinite);
    }
    
    /**
     * Retrieves messages from the server (Java ~ GetResponses).
     * 
     * <p>Works by pulling a default maximum of 10,000 messages from the server at a time.
     * If there are more messages on the server, it will continue to pull until all messages have been retrieved
     * or a timeout has occurred.</p>
     *  
     * @param <TMessage> Type of message to retrieve
     * @param responseClass Type of message to retrieve
     * @param timeoutMilliseconds Timeout in milliseconds. Increase this amount or decrease numMsgsPull if timeouts keep occurring
     * @return a list of BrokerResponse classes
     * @see BrokerResponse
     * @throws TimeoutException if pullResponses is unable to complete before timing out
     * @throws SessionException if there was a session error
     */
    public <TMessage> List<BrokerResponse<TMessage>> pullResponses(Class<TMessage> responseClass, long timeoutMilliseconds) throws TimeoutException, SessionException
    {
        String responseAction = wsdlFile.getResponseAction(responseClass.getSimpleName());
        Timeout custom = Timeout.Custom;
        custom.setTimeout(timeoutMilliseconds);
        return pullResponses(responseClass, responseAction, sendCount, custom);
    }
    
    /**
     * Retrieves messages from the server (Java ~ GetResponses).
     * 
     * <p>Works by pulling a default maximum of 10,000 messages from the server at a time.
     * If there are more messages on the server, it will continue to pull until all messages have been retrieved
     * or a timeout has occurred.</p>
     *  
     * @param <TMessage> Type of message to retrieve
     * @param responseClass Type of message to retrieve
     * @param numMsgPull Number of messages to try to pull at one time. Reduce this amount or increase the timeout if timeouts keep occurring.
     * @return a list of BrokerResponse classes
     * @throws TimeoutException  if pullResponses is unable to complete before timing out
     * @throws SessionException  if there was a session error
     */
    public <TMessage> List<BrokerResponse<TMessage>> pullResponses(Class<TMessage> responseClass, int numMsgPull) throws TimeoutException, SessionException
    {
        String responseAction = wsdlFile.getResponseAction(responseClass.getSimpleName());
        return pullResponses(responseClass, responseAction, numMsgPull, Timeout.Infinite);
    } 
 
    /**
     * Retrieves messages from the server (Java ~ GetResponses).
     * 
     * <p>Works by pulling a default maximum of 10,000 messages from the server at a time.
     * If there are more messages on the server, it will continue to pull until all messages have been retrieved
     * or a timeout has occurred.</p>
     * 
     * @param <TMessage> Type of message to retrieve
     * @param responseClass Type of message to retrieve
     * @param numMsgPull Number of messages to try to pull at one time. Reduce this amount or increase the timeout if timeouts keep occurring.
     * @return
     * @throws TimeoutException 
     * @throws SessionException 
     */
    public <TMessage> List<BrokerResponse<TMessage>> pullResponses(Class<TMessage> responseClass, int numMsgPull, long timeoutMilliseconds) throws TimeoutException, SessionException
    {
        String responseAction = wsdlFile.getResponseAction(responseClass.getSimpleName());
        Timeout custom = Timeout.Custom;
        custom.setTimeout(timeoutMilliseconds);
        return pullResponses(responseClass, responseAction, numMsgPull, custom);
    } 
    
    /**
     * Retrieves messages from the server (Java ~ GetResponses)
     * @param <TMessage> Type of message to retrieve
     * @param responseClass Type of message to retrieve
     * @param action Action EPR
     * @return
     * @throws TimeoutException 
     */
    public <TMessage> List<BrokerResponse<TMessage>> pullResponses(Class<TMessage> responseClass, String responseAction) throws TimeoutException, SessionException
    {
        return pullResponses(responseClass, responseAction, sendCount, Timeout.Infinite);
    }
    
    /**
     * Retrieves messages from the server (Java ~ GetResponses)
     * @param <TMessage> Type of message to retrieve
     * @param responseClass Type of message to retrieve
     * @param action Action EPR
     * @return
     * @throws TimeoutException 
     */
    public <TMessage> List<BrokerResponse<TMessage>> pullResponses(Class<TMessage> responseClass, String responseAction, long timeoutMilliseconds) throws TimeoutException, SessionException
    {
        Timeout custom = Timeout.Custom;
        custom.setTimeout(timeoutMilliseconds);
        return pullResponses(responseClass, responseAction, sendCount, custom);
    }
    
    /**
     * Retrieves messages from the server (Java ~ GetResponses)
     * @param <TMessage> Type of message to retrieve
     * @param responseClass Type of message to retrieve
     * @param action Action EPR
     * @param numMsgPull Number of messages to try to pull at one time (WARNING: may cause a timeout if try to pull too many messages)
     * @return
     * @throws TimeoutException 
     */
    public <TMessage> List<BrokerResponse<TMessage>> pullResponses(Class<TMessage> responseClass, String responseAction, int numMsgPull) throws TimeoutException, SessionException
    {
        return pullResponses(responseClass, responseAction, numMsgPull, Timeout.Infinite);
    } 
    
    /**
     * Retrieves messages from the server (Java ~ GetResponses)
     * @param <TMessage> Type of message to retrieve
     * @param responseClass Type of message to retrieve
     * @param action Action EPR
     * @param numMsgPull Number of messages to try to pull at one time
     * @param timeout Timeout enum containing timeout values
     * @see Timeout
     * @return
     */
    public <TMessage> List<BrokerResponse<TMessage>> pullResponses(Class<TMessage> responseClass, String responseAction, int numMsgPull, Timeout timeout)
    throws TimeoutException, SessionException
    {
        //Date targetTimeout = new Date(System.currentTimeMillis() + timeout.getTimeout());
        
        Boolean isEOM = false;

        int fetchCount = 0;
        
        
        List<BrokerResponse<TMessage>> list = new LinkedList<BrokerResponse<TMessage>>();
        
        IBrokerController controllerClient = getControllerClient();
        
        int batchCount = (numMsgPull > MAX_INITIAL_BATCH_COUNT || numMsgPull == 0) ? MAX_INITIAL_BATCH_COUNT : numMsgPull;
        // Loop until EOM
        while(!isEOM)
        {
            //setTimeout(controllerClient, getTimeout(targetTimeout));
            
            BrokerResponseMessages responses = null;
            
            GetResponsePosition position = (fetchCount == 0) ? GetResponsePosition.BEGIN : GetResponsePosition.CURRENT;
            
            responses = controllerClient.pullResponses(responseAction, 
                    position, batchCount, clientId);
            
            fetchCount++;
            numMsgPull -= batchCount;
            // Loop through response batch
            
            
            for(BrokerResponseMessages.SOAPMessage response : responses.getSOAPMessage())
            {
                // Convert the message XML to a SOAPMessage
                list.add(new BrokerResponse<TMessage>(responseClass, getSOAPMessage(response.getAny())));
            }
            
            isEOM = responses.isEOM();
            
            // Pause between requesting response batches
            if(!isEOM)
            {
                try {
                    Thread.sleep(REQUEST_RESPONSES_INTERVAL);
                }
                
                catch(Exception e) {
                    break;
                }
            }
        }
        return list;
    }

    /**
     * Gets the message out of a SOAP xml file.
     * @param messageXml message to be parsed
     * @return
     */
    public SOAPMessage getSOAPMessage(Object messageXml) throws SessionException
    {
        try {
        
            org.w3c.dom.Element element = (org.w3c.dom.Element) messageXml;
            
            // Convert message from Element to string
            TransformerFactory tf = TransformerFactory.newInstance();
            Transformer trans = tf.newTransformer();
            StringWriter sw = new StringWriter();
            trans.transform(new DOMSource(element), new StreamResult(sw));
            InputStream stream = new ByteArrayInputStream(sw.toString().getBytes());
            
            // Convert message from string to SOAPMessage
            MessageFactory mf = MessageFactory.newInstance();
            return mf.createMessage(new MimeHeaders(), stream);
        }
        
        catch(Exception e) {
            throw new SessionException(e);
        }

    }
    
    /**
     * Creates a dispatch which forwards the requests to the server
     * @param service service name 
     * @param portName binding
     * @param address Broker endpoint address
     * @param action Action string
     * @return
     */
    private Dispatch<SOAPMessage> getSOAPDispatch(TContract service, Binding binding, String address, String action)
    {   
        if(dispatchMap.containsKey(action))
        {
            return dispatchMap.get(action);
        }
        else
        {
            Dispatch<SOAPMessage> dispatch = 
                //binding.getBindingInfo should be enough to tell the cluster which binding to use to accomplish the request
                service.createDispatch(binding.getBindingInfo().getName(), SOAPMessage.class, Service.Mode.MESSAGE, new AddressingFeature(true));
            dispatch.getRequestContext().put(Dispatch.ENDPOINT_ADDRESS_PROPERTY, address);
            dispatch.getRequestContext().put(Dispatch.SOAPACTION_USE_PROPERTY, true);
            dispatch.getRequestContext().put(Dispatch.SOAPACTION_URI_PROPERTY, action);
            dispatchMap.put(action, dispatch);
            return dispatch;
        }
    }
    
    /**
     * Generates the SOAP message for the dispatch
     * @param payload the message to be generated
     * @param userData userdata to be supplied with the message
     * @return
     */
    private SOAPMessage createSOAPMessage(Object payload, Object userData) throws SessionException
    {
        try {
            // Create message factory
            MessageFactory mf = MessageFactory.newInstance();
            SOAPMessage msg = mf.createMessage();
            
            // If the caller specified userData, add it as a SOAP header
            if(userData != null)
            {
                generateUserDataHeader(userData, msg);                                                                       
            }
            setClientIdHeader(msg);
            
            // Add the payload object as a message body
            JAXBContext jc = JAXBContext.newInstance(payload.getClass());
            Marshaller marshaller = jc.createMarshaller();
            marshaller.marshal(payload, msg.getSOAPBody());
            msg.saveChanges();
            
            return msg;
        }       
        catch(Exception e) {
            throw new SessionException(e);
        }
    }
    
    /**
     * Sets the HTTP client proxy timeout
     * There is an error with this function as of build 3.0.1966 
     * @param serviceObject
     * @param timeout
     */
    @SuppressWarnings("unused")
    private void setTimeout(Object serviceObject, long timeout)
    {
        Client client = ClientProxy.getClient(serviceObject);
        
        HTTPConduit http = (HTTPConduit) client.getConduit();
        
        HTTPClientPolicy httpClientPolicy = http.getClient();
        
        httpClientPolicy.setConnectionTimeout(timeout);
        httpClientPolicy.setReceiveTimeout(timeout);
        
        http.setClient(httpClientPolicy);
    }
    
    /**
     * Get the time span from the targeted time
     * @param targetTimeout
     * @return
     * @throws TimeoutException
     */
    @SuppressWarnings("unused")
    private long getTimeout(Date targetTimeout) throws TimeoutException
    {
        Date currentTime = new Date();
        if(targetTimeout.after(currentTime))
        {
            return targetTimeout.getTime() - currentTime.getTime();
        }
        else
        {
            throw new TimeoutException();
        }
    }
    
    /**
     * Set SOAP header for clientId
     * <p>Java Changes: return type to void, added SOAPMessage msg as parameter
     * Required for interoperability with Java's native SOAP functions</p>
     * @param msg
     * @throws SOAPException
     */
    private void setClientIdHeader(SOAPMessage msg) throws SOAPException
    {
        SOAPHeader header = msg.getSOAPHeader();
        SOAPHeaderElement headerElement = header.addHeaderElement(
                    clientIdHeader);
        headerElement.setValue(clientId.toString());
    }
   
    
}
