//
// FlowProcessor.java - Flow Processor
//
// Queues and Topics sit in a unified Destination namespace, and because some
// JMS implementations allow the same name for a Queue and a Topic, at the same
// time, Queues have "Q" prepended to their names, Topics have "T".
//
// Static mappings are populated from the configuration file.
//
// If an outbound message has a reply flow and a JMSReplyTo, we remember
// to transform inbound replies which arrive on the reply destination.
// Proliferation of these dynamically provisioned inbound mappings doesn't
// occur, because if the reply destination is a TemporaryQueue or
// TemporaryTopic, we get notified when they are deleted.
//
// If an inbound message has a reply flow and a JMSReplyTo, we remember
// to transform outbound replies which we send to the reply destination.
// The reply destinations can be temporary, but we cannot tell this, and we
// don't know when they cease to exist. So we keep a LRU cache of limited size.
//

package nyangau.jmsadapter;

//...simports:0:
import java.util.List;
import java.util.LinkedList;
import java.util.Map;
import java.util.HashMap;
import java.util.Properties;
import java.io.*;
import java.net.URL;
import javax.jms.*;
import javax.xml.XMLConstants;
import javax.xml.namespace.NamespaceContext;
import javax.xml.parsers.DocumentBuilder;
import javax.xml.parsers.DocumentBuilderFactory;
import javax.xml.parsers.ParserConfigurationException;
import javax.xml.validation.SchemaFactory;
import javax.xml.validation.Schema;
import javax.xml.validation.Validator;
import javax.xml.transform.stream.StreamSource;
import javax.xml.transform.dom.DOMSource;
import org.xml.sax.InputSource;
import org.xml.sax.ErrorHandler;
import org.xml.sax.SAXException;
import org.xml.sax.SAXParseException;
import org.w3c.dom.Document;
import org.w3c.dom.Element;
import org.w3c.dom.NodeList;
import org.w3c.dom.NamedNodeMap;
import org.w3c.dom.Node;
import org.w3c.dom.Text;
//...e

public class FlowProcessor
  {
  protected Map<String,Object> p;
  protected Map<String,Flow> flows;
  protected Map<String,DataMap> datamaps;
//...sgetFlow:2:
// Lookup flow by name

public Flow getFlow(String name)
  { return flows.get(name); }
//...e
//...sgetDataMap:2:
// Lookup datamap by name

public DataMap getDataMap(String name)
  { return datamaps.get(name); }
//...e
//...sclass FlowAndReply:2:
public class FlowAndReply
  {
  public Flow flow;
  public Flow flowReply;
  public FlowAndReply(Flow f, Flow fr)
    { flow = f; flowReply = fr; }  
  }
//...e
//...smappings from destinations to FlowAndReplys:2:
protected Object mapSync = new Object();
// All outbound mappings, static or dynamic
protected Map<String,FlowAndReply> farsO = new HashMap<String,FlowAndReply>();
// All inbound mappings, static or dynamic
protected Map<String,FlowAndReply> farsI = new HashMap<String,FlowAndReply>();
// LRU cache of dynamically added outbound mappings
protected int cacheSize = 1000;
protected Map<String,FlowAndReply> farsOdynH = new HashMap<String,FlowAndReply>();
protected List<String> farsOdynL = new LinkedList<String>(); // Most recent inbound added replies
//...e
//...sunifiedDestName:2:
protected String unifiedDestName(Destination d)
  throws JMSException
  {
  if ( d instanceof Queue )
    return "Q"+((Queue)d).getQueueName();
  else if ( d instanceof Topic )
    return "T"+((Topic)d).getTopicName();
  else if ( d == null )
    throw new JMSException("null Destination");
  else
    throw new JMSException("weird Destination");
  }
//...e
//...sFlowProcessor:2:
// This is where we pay the price of parsing the XML configuration file,
// substituting properties and constructing flows.

//...snewDestination:2:
//...snewFlowRef:2:
protected Flow newFlowRef(NamedNodeMap nnm, String attr)
  throws FlowProcessorException
  {
  Node a = nnm.getNamedItem(attr);
  if ( a == null )
    return null;
  String n = a.getNodeValue();
  Flow f = getFlow(n);
  if ( f == null )
    throw new FlowProcessorException("unexpected reference to non-exsitent flow, bad jmsadapter.xsd?");
  return f;
  }
//...e

protected void newDestination(String namePrefix, Node eDest)
  throws FlowProcessorException
  {
  NamedNodeMap nnmDest = eDest.getAttributes();
  Node aName = nnmDest.getNamedItem("name");
  if ( aName == null )
    throw new FlowProcessorException("unexpected missing name in queue or topic, bad jmsadapter.xsd?");
  String name = aName.getNodeValue();
  Flow flowOut      = newFlowRef(nnmDest, "out"     );
  Flow flowOutReply = newFlowRef(nnmDest, "outreply");
  if ( flowOut != null || flowOutReply != null )
    farsO.put(namePrefix+name, new FlowAndReply(flowOut, flowOutReply));
  Flow flowIn      = newFlowRef(nnmDest, "in"     );
  Flow flowInReply = newFlowRef(nnmDest, "inreply");
  if ( flowIn != null || flowInReply != null )
    farsI.put(namePrefix+name, new FlowAndReply(flowIn, flowInReply));
  }
//...e

public FlowProcessor(Map<String,Object> map)
  throws FlowProcessorException
  {
  p = map; // keep it

  String configUrl = (String) p.get(Jmsa.CONFIG);
  if ( configUrl == null )
    throw new FlowProcessorException(Jmsa.CONFIG + " property not set");
  String config;
  try
    {
    config = FlowHelper.fetchURL(configUrl);
    }
  catch ( IOException e )
    {
    throw new FlowProcessorException("can't fetch configuration from "+configUrl, e);
    }

  Properties properties = new Properties();
  String propertiesUrl = (String) p.get(Jmsa.PROPERTIES);
  if ( propertiesUrl != null )
    {
    try
      {
      InputStream is2 = new URL(propertiesUrl).openStream();
      properties.load(is2);
      }
    catch ( IOException e )
      {
      throw new FlowProcessorException ("can't fetch properties from "+propertiesUrl, e);
      }
    }
  config = FlowHelper.substituteProps(config, properties);

  DocumentBuilderFactory dbf = DocumentBuilderFactory.newInstance();
  // not validating, we'll do that below
  dbf.setNamespaceAware(true);
  dbf.setCoalescing(true);
  dbf.setExpandEntityReferences(true);
  dbf.setIgnoringComments(true);
  dbf.setIgnoringElementContentWhitespace(false);
  Document d;
  try
    {
    DocumentBuilder db = dbf.newDocumentBuilder();
    d = db.parse(new InputSource(new StringReader(config)));
      // Note: we avoid using new InputStream(new StringInputStream()),
      // as we want to avoid working at the byte level,
      // after all, we already have a String, which is a sequence of chars.
      // Note that StringInputStream is deprecated for this reason.
    d.normalize();
    }
  catch ( ParserConfigurationException e )
    {
    // Serious problems
    throw new FlowProcessorException("can't parse configuration", e);
    }
  catch ( SAXParseException e )
    {
    throw new FlowProcessorException("can't parse configuration, line "+e.getLineNumber()+", uri "+e.getSystemId(), e);
    }
  catch ( SAXException e )
    {
    throw new FlowProcessorException("can't parse configuration", e);
    }
  catch ( IOException e )
    {
    // What are the odds of this, our InputStream is in memory!
    throw new FlowProcessorException("unexpected IOException during parsing, configuration is in memory", e);
    }

  Schema s;
  try
    {
    SchemaFactory sf = SchemaFactory.newInstance(XMLConstants.W3C_XML_SCHEMA_NS_URI);
    InputStream is = this.getClass().getResourceAsStream("/jmsadapter.xsd");
    s = sf.newSchema(new StreamSource(is));
    }
  catch ( SAXException e )
    // We've ensured jmsadapter.xsd parses cleanly in advance, so this won't happen
    {
    throw new FlowProcessorException("unexpected SAXException, bad jmsadapter.xsd?");
    }

  try
    {
    Validator v = s.newValidator();
    v.validate(new DOMSource(d));
    }
  catch ( SAXException e )
    // So the users configuration file isn't valid
    {
    throw new FlowProcessorException("configuration is not valid", e);
    }
  catch ( IOException e )
    // d is in memory, so this isn't going to happen
    {
    throw new FlowProcessorException("unexpected IOException during validation, configuration is in memory", e);
    }

  datamaps = new HashMap<String,DataMap>();
  NodeList nl = d.getDocumentElement().getChildNodes();
  for ( int i = 0; i < nl.getLength(); i++ )
    {
    Node e = nl.item(i);
    if ( e.getNodeType() == Node.ELEMENT_NODE &&
         e.getNodeName().startsWith("map") )
      {
      DataMap datamap = DataMap.newDataMap(e);
      datamaps.put(datamap.getName(), datamap);
      }
    }

  flows = new HashMap<String,Flow>();
  NodeList nlFlows  = d.getElementsByTagName("flow");
  for ( int i = 0; i < nlFlows.getLength(); i++ )
    {
    Node eFlow = nlFlows.item(i);
    Flow f = new Flow(eFlow);
    flows.put(f.getName(), f);
    }

  NodeList nlQueues = d.getElementsByTagName("queue");
  for ( int i = 0; i < nlQueues.getLength(); i++ )
    {
    Node eQueue = nlQueues.item(i);
    newDestination("Q", eQueue);
    }

  NodeList nlTopics = d.getElementsByTagName("topic");
  for ( int i = 0; i < nlTopics.getLength(); i++ )
    {
    Node eTopic = nlTopics.item(i);
    newDestination("T", eTopic);
    }

  }
//...e
//...sprocess:2:
// Overall exception strategy is
//   try to execute flow, and if no exception occurs, then done
//   if there is an exception flow defined
//     set _JMSAdapter_flow_context, _name and _exception variables
//     hope the flow has set up the _JMSAdapter_message_id variable
//     try to execute exception flow, and if no exception occurs, then done
//     zap the message to give an empty JmsaMessage
//     set _JMSAdapter_eflow_context, _name and _exception properties
//     copy _JMSAdapter_flow_context, _name and _exception vars to properties
//     if _JMSAdapter_message_id String variable defined, copy to property
//     hope the receiver will choke nicely on this message   :-)
//   else
//     zap the message to an empty JmsaMessage
//     copy _JMSAdapter_flow_context, _name and _exception vars to properties
//     if _JMSAdapter_message_id String variable defined, copy to property
//     hope the receiver will choke nicely on this message   :-)

//...sprocessException:2:
//...sprocessException2:2:
//...scopyVarToProp:2:
protected static void copyVarToProp(FlowContext c, String name)
  throws JMSException
  {
  Object o = c.getVar(name);
  if ( o != null )
    {
    JmsaMessage m = c.getMessage();
    if ( o instanceof String )
      m.setStringProperty(name, (String) o);
    else if ( o instanceof Exception )
      m.setStringProperty(name, FlowHelper.diagForm((Exception)o));
    }
  }
//...e

protected void processException2(Flow ef, Exception ee, FlowContext c)
  // DOES NOT throw FlowProcessorException, JMSException
  {
  try
    {
    JmsaMessage m = c.getMessage();
    m = new JmsaMessage(m); // convert to raw message with no body
    c.setMessage(m);
    m.clearProperties(); // wipe out any properties
    if ( ef != null )
      {
      m.setStringProperty("_JMSAdapter_eflow_context", FlowHelper.diagForm(c, true, true, null));
      m.setStringProperty("_JMSAdapter_eflow_name", ef.getName());
      m.setStringProperty("_JMSAdapter_eflow_exception", FlowHelper.diagForm(ee));
      }
    copyVarToProp(c, "_JMSAdapter_flow_context");
    copyVarToProp(c, "_JMSAdapter_flow_name");
    copyVarToProp(c, "_JMSAdapter_flow_exception");
    copyVarToProp(c, "_JMSAdapter_message_id");
    }
  catch ( JMSException e2 )
    {
    // isn't going to happen, we're just manipulating memory
    }
  }
//...e

protected void processException(Flow f, Exception e, FlowContext c)
  // DOES NOT throw FlowProcessorException, JMSException
  {
  // Make these available to exception flow
  try
    {
    c.setVar("_JMSAdapter_flow_context", FlowHelper.diagForm(c, true, true, null));
    }
  catch ( JMSException e2 )
    {
    // isn't going to happen, we're just manipulating memory
    c.setVar("_JMSAdapter_flow_context", "");
    }
  c.setVar("_JMSAdapter_flow_name", f.getName());
  c.setVar("_JMSAdapter_flow_exception", e);
  String eflow = f.getEflow();
  Flow ef;
  if ( eflow != null && (ef = getFlow(eflow)) != null )
    // Run the exception flow
    try
      {
      ef.process(c);
      }
    catch ( FlowProcessorException ee )
      {
      processException2(ef, ee, c);
      }
    catch ( JMSException ee )
      {
      processException2(ef, ee, c);
      }
    catch ( RuntimeException ee )
      {
      processException2(ef, ee, c);
      }
  else
    processException2(null, null, c);
  }
//...e

public void process(Flow f, FlowContext c)
  // DOES NOT throw FlowProcessorException, JMSException
  {
  try
    {
    f.process(c);
    }
  catch ( FlowProcessorException e )
    {
    processException(f, e, c);
    }
  catch ( JMSException e )
    {
    processException(f, e, c);
    }
  catch ( RuntimeException e )
    {
    processException(f, e, c);
    }
  }

protected JmsaMessage process(JmsaMessage m, Flow f)
  // DOES NOT throw FlowProcessorException, JMSException
  {
  FlowContext c = new FlowContext(this, m);
  process(f, c);
  return c.getMessage();
  }
//...e
//...sprocessOut:2:
// At this time the JMSDestination is not valid, but JMSReplyTo is

public JmsaMessage processOut(Destination d, JmsaMessage m)
  throws JMSException
  {
  Destination dr = m.getJMSReplyTo();
  String dName = unifiedDestName(d);
  FlowAndReply farO;
  synchronized ( mapSync )
    {
    farO = farsO.get(dName);
    }
  if ( farO != null )
    // Transform on the way out
    {
    if ( farO.flow != null )
      {
      m = JmsaMessage.copyOfMessage(m);
      m = process(m, farO.flow);
      }
    synchronized ( mapSync )
      {
      if ( farsOdynH.get(dName) != null )
        // Remember we recently touched this
        {
        farsOdynL.remove(dName); // Slow, linked list search
        farsOdynL.add(dName);
        }
      if ( farO.flowReply != null && dr != null )
        // Remember we'd better transform on the way back in
        {
        String drName = unifiedDestName(dr);
        FlowAndReply farI = farsI.get(drName);
        if ( farI != null )
          farI.flow = farO.flowReply;
        else
          {
          farI = new FlowAndReply(farO.flowReply, null);
          farsI.put(drName, farI);
          } 
        }
      }
    }
  return m;
  }
//...e
//...sprocessIn:2:
public JmsaMessage processIn(JmsaMessage m)
  throws JMSException
  {
  Destination d  = m.getJMSDestination();
  Destination dr = m.getJMSReplyTo();
  String dName = unifiedDestName(d);
  FlowAndReply farI;
  synchronized ( mapSync )
    {
    farI = farsI.get(dName);
    }
  if ( farI != null )
    // Transform on the way in
    {
    if ( farI.flow != null )
      {
      m = JmsaMessage.copyOfMessageWithDestinations(m);
      m = process(m, farI.flow);
      }
    synchronized ( mapSync )
      {
      if ( farI.flowReply != null && dr != null )
        // Remember we'd better transform on the way back out
        {
        String drName = unifiedDestName(dr);
        FlowAndReply farO = farsO.get(drName);
        if ( farO != null )
          {
          farO.flow = farI.flowReply;
          if ( farsOdynH.get(drName) != null )
            // Remember we recently touched this
            {
            farsOdynL.remove(dName);
            farsOdynL.add(dName);
            }
          }      
        else
          // We assume the JMSReplyTo identifies a temporary destination
          // and unfortunately, we can't know when this destination ceases
          // to exist, so we keep a LRU cache of limited size.
          {
          farO = new FlowAndReply(farI.flowReply, null);
          farsO.put(drName, farO);
          farsOdynH.put(drName, farO);
          farsOdynL.add(drName);
          if ( farsOdynL.size() > cacheSize )
            // Discard the LRU entry
            {
            String dForget = farsOdynL.get(0);
            farsO.remove(dForget);
            farsOdynH.remove(dForget);
            farsOdynL.remove(0);
            }
          }
        }
      }
    }
  return m;
  }
//...e
//...sforgetTemporary:2:
protected void forgetTemporary(String name)
  {
  synchronized ( mapSync )
    {
    farsI.remove(name);
    }
  }
//...e
//...sforgetTemporaryQueue:2:
public void forgetTemporaryQueue(String qn)
  { forgetTemporary("Q"+qn); }
//...e
//...sforgetTemporaryTopic:2:
public void forgetTemporaryTopic(String tn)
  { forgetTemporary("T"+tn); }
//...e
  }
