C# .NET client to publish 5 million events to Universal Messaging channel\IS Publishable document

In this blog article I have created a C# .NET client using Universal Messaging C# .NET API. This is in contrast to my previous two articles where I used Java and Message builders(as described in IBM official documentation) and Dynamic Builders to compose a Protobuf event and publish large amount of data to UM Channel\IS Publishable document. I took on the effort to do the same thing in C# to see if there will be a performance difference or if there is a different way to send messages to UM Channel more efficiently and to my surprise the performance metrics are so different between both implementations. I have described more in detail about the performance differences with same UM and IS setup in detail in the blog. Generate the C# code from the proto file. Below command is for running it from the same folder where the proto file is located. protoc --csharp_out=. Sales_pdt.proto

This will generate a new cs file called SalesPdt.cs as show below.

At this point we have the auto generated csharp file generaed by the protoc compiler that we can import in to our existing solution and generate a nProtobufevent from the data and then loop over each generated event and publish to UM.

I have listed out a step-by-step instructions on how to implement a C# .NET program to generate a proto buf event and publish the event to UM. You can download the pdf document here.

For any questions or inquireies feel free to reach out to aarremreddy@silverspringtech.com

Program.cs

using com.pcbsys.nirvana.client; using Google.Protobuf.Reflection; using Google.Protobuf; using System.Text; using System.Reflection; using MyChannels.Nirvana; using IBM_MQ; nSession mySession = UMUtilities.UMConnect("nsp://localhost:9000"); nChannel myChannel = UMUtilities.setChannel("wm/is/test/Akshith/docs/SalesDataDoc", mySession); String[] records=File.ReadAllLines("C:\\Users\\test\\Downloads\\5m-Sales-Records\\5m Sales Records.csv"); Console.WriteLine(DateTime.Now+" -> Start publishing documents to UM."); for(int i=0; i<<records.Length; i++){ myChannel.publish(Sales.generateSalesEvents(records[i])); } mySession.close(); Console.WriteLine(DateTime.Now + " -> Finished publishing documents to UM."); Console.ReadLine();

Sales.cs

using com.pcbsys.nirvana.client; using Google.Protobuf; using System; using System.Collections.Generic; using System.Linq; using System.Text; using System.Threading.Tasks; namespace IBM_MQ { class Sales { public static nProtobufEvent generateSalesEvents(String inRecord) { String[] fields = inRecord.Split(','); dli_test_Akshith_docs_SalesDataDoc outDoc=new dli_test_Akshith_docs_SalesDataDoc(); dli_test_Akshith_docs_SalesDataDoc_salesData salesData = new dli_test_Akshith_docs_SalesDataDoc_salesData(); salesData.Region = fields[0]; salesData.Country = fields[1]; salesData.ItemType = fields[2]; salesData.SalesChannel = fields[3]; salesData.OrderPriority = fields[4]; salesData.OrderDate = fields[5]; salesData.OrderId = fields[6]; salesData.ShipDate = fields[7]; salesData.UnitsSold = fields[8]; salesData.UnitPrice = fields[9]; salesData.UnitCost = fields[10]; salesData.TotalRevenue = fields[11]; salesData.TotalCost = fields[12]; salesData.TotalProfit = fields[13]; outDoc.SalesData = salesData; Sales_pub_publish_envelope env=new Sales_pub_publish_envelope(); env.TransactionId = "asddfad"; outDoc.Env = env; nProtobufEvent evt = new nProtobufEvent(outDoc.ToByteArray(),"salesData"); return evt; } } }

Java client to publish 5 million events to Universal Messaging channel\IS Publishable document

This is a continuation to the previous blog where I have shown how to publish a protobuf event from Java to IBM Universal Messaging channel which is also a publishable document on IBM webMethods Integration Server.

This whole process took about 4 minutes to publish 5 million records to the UM channel with the specifications that I have described in specification section. On processing end, the processing time for 5 million records will depend on the number of threads you have configured on the webMethods messaging trigger and the number of wM Integration servers in a cluster. I have tested this with a 2 node webMethods Integration Server cluster, and it took about 5 minutes to process the 5 million records.

The more webMethods Integration Servers you have in the cluster and more the number of threads working in parallel processing the messages faster.

You can Download the PDF with detailed step-by-step instructions along with the full code.

Note: Feel free to reach out to aarremreddy@silverspringtech.com for any questions or inquiries.

Proto file: message test_Akshith_docs_SalesDataDoc { optional test_Akshith_docs_SalesDataDoc_salesData salesData = 1; optional pub_publish_envelope _env = 536870910; } message test_Akshith_docs_SalesDataDoc_salesData { optional string region = 1; optional string country = 2; optional string itemType = 3; optional string salesChannel = 4; optional string orderPriority = 5; optional string orderDate = 6; optional string orderId = 7; optional string shipDate = 8; optional string unitsSold = 9; optional string unitPrice = 10; optional string unitCost = 11; optional string totalRevenue = 12; optional string totalCost = 13; optional string totalProfit = 14; } message pub_publish_envelope { optional string activation = 1; optional int32 appLastSeqn = 2; optional string appPassword = 3; optional int32 appSeqn = 4; optional string appUserName = 5; optional int32 priority = 6; optional string businessContext = 7; repeated int32 controlLabel = 8; optional string errorsTo = 9; optional string errorRequestsTo = 10; optional string locale = 11; optional int32 maxResults = 12; optional string replyTo = 13; optional int32 runLevel = 14; optional bytes signature = 15; optional string signatureType = 16; optional int32 startResult = 17; optional int32 tag = 18; optional string trackId = 19; optional string transactionId = 20; optional string transformState = 21; optional int32 age = 22; optional string connectionIntegrity = 23; optional string destId = 24; optional int64 enqueueTime = 25; optional string logBroker = 26; optional string logHost = 27; optional string pubDistinguishedName = 28; optional string pubId = 29; optional bytes pubNetAddr = 30; optional int64 pubSeqn = 31; repeated int32 pubLabel = 32; optional int64 recvTime = 33; repeated pub_publish_envelope_route route = 34; optional string uuid = 35; optional string status = 36; } message pub_publish_envelope_route { optional string broker = 1; optional int64 recvTime = 2; optional int64 enqueueTime = 3; } //Read File Line By Line while ((strLine = br.readLine()) != null) { tempData=new SalesData(); String[] tempList=strLine.split(","); tempData.setRegion(tempList[0]); tempData.setCountry(tempList[1]); tempData.setItemType(tempList[2]); tempData.setSalesChannel(tempList[3]); tempData.setOrderPriority(tempList[3]); tempData.setOrderDate(tempList[4]); tempData.setOrderId(tempList[5]); tempData.setShipDate(tempList[6]); tempData.setUnitsSold(tempList[7]); tempData.setUnitPrice(tempList[8]); tempData.setUnitCost(tempList[9]); tempData.setTotalRevenue(tempList[10]); tempData.setTotalCost(tempList[11]); tempData.setTotalProfit(tempList[12]); nProtobufEvent evt=UMUtilities.SalesDataToProtoEvt(descriptorFileBytes,tempData); myChannel.publish(evt); evt=null; }

Java client to publish Protbuf events to Universal Messaging channel\IS Publishable document

In this topic I will be sharing the code to send ProtoBuf events from a standalone Java client to Universal Messaging Channel which is also a publishable document on the webMethods Integration Server using the UM Native Java API.

By having this capability Companies can use existing products to achieve a seamless integration  between J2EE applications and webMethods without spending additional time and money on investing in new tools and infrastructure.

Also attached to this post is a detailed PDF documentation on how to implement the overall solution step by step. 

Creating a publishable document with encoding type set to "Protocol buffers" will create a proto file on the file system where the publishable doc is stored. The contents of the publishable doc that i used for this use case looks like below.

Note:

If you want to skip over this blog and look at the step by step instructions on how to implement this then you can download the PDF Documentation for this implementation. For any questions or if you are a client looking to request a demo feel free to reach out to aarremreddy@silverspringtech.com

Please note that this exercise is to demo the ability of publishing events to UM and does not include async error listeners or any other complex error handling scenarios.

Proto File content:

message test_Akshith_docs_testPubDoc { optional string message = 1; optional pub_publish_envelope _env = 536870910; } message pub_publish_envelope { optional string activation = 1; optional int32 appLastSeqn = 2; optional string appPassword = 3; optional int32 appSeqn = 4; optional string appUserName = 5; optional int32 priority = 6; optional string businessContext = 7; repeated int32 controlLabel = 8; optional string errorsTo = 9; optional string errorRequestsTo = 10; optional string locale = 11; optional int32 maxResults = 12; optional string replyTo = 13; optional int32 runLevel = 14; optional bytes signature = 15; optional string signatureType = 16; optional int32 startResult = 17; optional int32 tag = 18; optional string trackId = 19; optional string transactionId = 20; optional string transformState = 21; optional int32 age = 22; optional string connectionIntegrity = 23; optional string destId = 24; optional int64 enqueueTime = 25; optional string logBroker = 26; optional string logHost = 27; optional string pubDistinguishedName = 28; optional string pubId = 29; optional bytes pubNetAddr = 30; optional int64 pubSeqn = 31; repeated int32 pubLabel = 32; optional int64 recvTime = 33; repeated pub_publish_envelope_route route = 34; optional string uuid = 35; optional string status = 36; } message pub_publish_envelope_route { optional string broker = 1; optional int64 recvTime = 2; optional int64 enqueueTime = 3; }

As part of pre-requisites install Google Protocol buffers compiler. Refer to the PDF documentation for more information. After this installation, run below command to generate the descriptor file for the proto file.

protoc --descriptor_set_out=E:\Akshith\protoc\pdt.descriptor pdt.proto  

The variables nSession and nSessionAttributes are initialized as static since they are common properties that will be shared by the class objects.

protected static nSession mySession = null; protected static nSessionAttributes nsa = null;  

The first step is to connect to the UM realm. In this case, I have a local instance of UM that does not have basic auth or SSL enabled so i am just using the realm endpoint for connection.

nsa = new nSessionAttributes("nsp://localhost:9000"); //This will be the realm endpoint that we are connecting to on the default 9000 port.
System.out.println("Session Attributes created.");

//Create a nSession object from the nSessionAttributes that are created using the endpoint above without any auth. mySession = nSessionFactory.create(nsa); // This is the line that you have to use for UM instances with basic auth enabled. //mySession=nSessionFactory.create(nsa,“username”,“password”); //For UM instances where SSL auth is enabled, below two lines need to be used. //nas.setKeystore("path/to/keystore.jks", "keystorePassword"); //nsa.setTruststore("path/to/truststore.jks", "truststorePassword"); System.out.println("Session created."); mySession.init();// This will start the connection to the UM realm.  

Once the connection is established then we have to create the Protbuf event. Create a builder of instance com.google.protobuf.DescriptorProtos.FileDescriptorSet and load the descriptor file contents in to the builder. This is the first step to build the Protobuf event.

FileDescriptorSet.Builder builder = DescriptorProtos.FileDescriptorSet.newBuilder(); String fileName="E:\\Akshith\\protoc\\pdt.descriptor"; byte[] bytes = fFile.readAllBytes(new File(fileName)); builder.mergeFrom(bytes);

Create a filedescritor of instance com.google.protobuf.Descriptors.FileDescriptor from the builder that was generated above that holds the descriptor structure.

System.out.println("Start creating file descriptors."); com.google.protobuf.Descriptors.FileDescriptor fileDescriptor = com.google.protobuf.Descriptors.FileDescriptor.buildFrom(builder.getFile(0), new FileDescriptor[0]); System.out.println("Start creating test_Akshith_docs_testPubDoc descriptors.");

Create a descriptor of instance com.google.protobuf.Descriptors.Descriptor to get the message for "test_Akshith_docs_testPubDoc" in the descriptor file.

Descriptor test_Akshith_docs_testPubDoc=fileDescriptor.findMessageTypeByName("test_Akshith_docs_testPubDoc");

Create a builder for above descriptor and set the value for the field that needs to be published.

DynamicMessage.Builder testPubDocBuilder = DynamicMessage.newBuilder(test_Akshith_docs_testPubDoc). setField(test_Akshith_docs_testPubDoc.findFieldByName("message"), "This is a test for proto buffer"); //build the above defined builder testPubDocBuilder.build(); System.out.println("End creating test_Akshith_docs_testPubDoc descriptors.");

Repeat the same process as above to create a descriptor for envelope IData document since this is a mandatory field to populate by webMethods when publishing documents. See troubleshooting section in the pdf for more information. At least populate one field under envelope since it is mandatory.

Descriptor envDescriptor = fileDescriptor.findMessageTypeByName("pub_publish_envelope"); DynamicMessage.Builder envBuilder = DynamicMessage.newBuilder(envDescriptor) .setField(envDescriptor.findFieldByName("activation"), UMUtilities.generateRandomString()) .setField(envDescriptor.findFieldByName("recvTime"), UMUtilities.getInt64ValueForTimestamp()) .setField(envDescriptor.findFieldByName("uuid"), UMUtilities.generateRandomString()); System.out.println("Start creating UMMessageBuilder Builder.");

Create a builder for sending the message to UM and set the values

DynamicMessage.Builder UMMessageBuilder = DynamicMessage.newBuilder(test_Akshith_docs_testPubDoc) .setField(test_Akshith_docs_testPubDoc.findFieldByName("message"), "test this") .setField(test_Akshith_docs_testPubDoc.findFieldByName("_env"), envBuilder.build()); System.out.println("End creating UMMessageBuilder Builder.");

Create a dynamic message of instance com.google.protobuf.DynamicMessage with the UM Message builder that was created above to generate the final event of instance com.pcbsys.nirvana.client.nProtobufEvent that can be sent to UM.

DynamicMessage dynamicMessage = UMMessageBuilder.build(); nProtobufEvent evt=new nProtobufEvent(dynamicMessage.toByteArray(), UMMessageBuilder.getDescriptorForType().getName());

Set the channel name to send the message to. The exact channel name can be found on the webMethods publishable doc type property "Provider definition". Use this to set the channel name on Channel Attributes.

System.out.println("Start publshing to Channel."); nChannelAttributes nca = new nChannelAttributes(); nca.setName("wm/is/test/Akshith/docs/testPubDoc"); System.out.println("Finished setting the Channel name. Finding the channel");

Create an nChannel instance by running findChannel method. This will set the channel object to the one referenced in above code.

nChannel myChannel = mySession.findChannel(nca); System.out.println("Found the channel.");

At this point we have the protobuf event created from the descriptor file and the destination channel set. The event at this point needs to be published.

System.out.println("publshing to Channel."); myChannel.publish(evt); System.out.println("Finished publshing to Channel successfully.");

Standard catch block to handle any errors that happen while establishing a connection,creating a protobuf event or while finding the destnation channel.

catch (Exception ex) { System.out.println("Error : "+ex.toString()); System.exit(1); }

Standard finally block to close the session to UM realm.

finally { mySession.close(); System.out.println("Session closed"); }

Troubleshooting\Error scenarios:

Issue:
If the publishable doc type is not changed from default IData Encoding type to "Protocol buffers" then the standalone Java app will not error but on the consumer end, below error will be logged in the wM IS server log and message will not be processed.

ERROR:
2024-11-05 13:15:57 EST [ISS.0153.0089C] (tid=165624) Trigger test.Akshith.triggers:receiveJavaMessages failed because of message decoding exception: com.wm.app.b2b.server.dispatcher.exceptions.MessagingCoderException: [ISS.0153.9302] Protocol buffer coder cannot find descriptor in document type: test.Akshith.docs:testPubDoc. Message has been acknowledged to messaging provider.

Issue:
If there are issues connecting to the realm then below error will happpen. Check you realm connection details and try again.

Error : com.pcbsys.nirvana.client.nRealmUnreachableException: Realm is currently not reachable:Realm was still unreachable after max retry count - 0

Issue:
If you do not populate the envelope descriptor while sending the event to channel then it will not error out in Java app but on the consumer side you will notice below error in wM Integration Server. Descriptor envDescriptor = fileDescriptor.findMessageTypeByName("pub_publish_envelope"); DynamicMessage.Builder envBuilder = DynamicMessage.newBuilder(envDescriptor) .setField(envDescriptor.findFieldByName("activation"), UMUtilities.generateRandomString()) .setField(envDescriptor.findFieldByName("recvTime"), UMUtilities.getInt64ValueForTimestamp()) .setField(envDescriptor.findFieldByName("uuid"), UMUtilities.generateRandomString()); System.out.println("Start creating UMMessageBuilder Builder.");

Error : 2024-11-05 10:20:18 EST [ISS.0153.0041E] (tid=330770) webMethods Messaging Trigger test.Akshith.triggers:receiveJavaMessages failed preprocessing: java.lang.NullPointerException: Cannot invoke "com.wm.data.IData.getCursor()" because "env" is null

Issue:
If you do not populate the correct datatype while setting the values for fields then below error will occur in the Java app and it will go in to catch block. You can refer to the proto filet to get the correct data type for each field.

Error : java.lang.IllegalArgumentException: Wrong object type used with protocol message reflection.Field number: 33, field java type: LONG, value type: java.lang.String

Issue:
If the correct channel name is not provided then below error will occur. Check the "Provider definition" property on the publishable doc type on webMethods Integration server.

Error :com.pcbsys.nirvana.client.nChannelNotFoundException: Channel could not be found on the server:Find : /wmtest/is/test/Akshith/docs/testPubDoc

Using webMethods Java service (StAX Parser) to split a huge XML file in to multiple small files.

I did this same requirement using SAX parser in my previous post. The limitation using the SAX Parser was that if an XML was more than 300 MB, the service would run out of memory trying to split the xml. Because of the out of memory issues, i had to finally opt for StAX paraser since it is more efficient than the SAX or DOM (definitely not). I have tested this code with up to 2 GB files and it works with no issues. For a 1 GB file it took about 38 Mins to execute.

File:

<?xml version="1.0" encoding="UTF-8?>
<rootNode>
       <childNode>
               <ordernumber>12354</ordernumber>
       </childNode>
       <childNode>
               <ordernumber>12355</ordernumber>
       </childNode>
       <childNode>
               <ordernumber>12356</ordernumber>
       </childNode>
</rootNode>

The result should look like this:

file1:

<?xml version="1.0" encoding="UTF-8?>
<rootNode>
       <childNode>
               <ordernumber>12354</ordernumber>
       </childNode>
       <childNode>
               <ordernumber>12355</ordernumber>
       </childNode>
</rootNode>

file2:

<?xml version="1.0" encoding="UTF-8?>
<rootNode>
       <childNode>
               <ordernumber>12356</ordernumber>
       </childNode>
</rootNode>

Imports:

import java.io.*;
import javax.xml.namespace.QName;
import javax.xml.stream.*;
import javax.xml.stream.events.*;
import javax.xml.transform.stream.StreamSource;

Inputs: xmlFileAbsoluteName, rootNode, repeatingNode, splitFilesTargetDirectory, splitFilePrefixName, numOfRecordsPerFile.
Java Service Code:

IDataCursor pipelineCursor = pipeline.getCursor();
String	xmlFileAbsoluteName = IDataUtil.getString( pipelineCursor, "xmlFileAbsoluteName" );
String	rootNode = IDataUtil.getString( pipelineCursor, "rootNode" );
String	repeatingNode = IDataUtil.getString( pipelineCursor, "repeatingNode" );
String	splitFilesTargetDirectory = IDataUtil.getString( pipelineCursor, "splitFilesTargetDirectory" );
String	splitFilePrefixName = IDataUtil.getString( pipelineCursor, "splitFilePrefixName" );
String	numOfRecordsPerFile = IDataUtil.getString( pipelineCursor, "numOfRecordsPerFile" );		
pipelineCursor.destroy();
		
writeToApplicationLogFile("START: Start executing the Java code to split large XML to small XML files.", logFileName, project);
XMLInputFactory inputFactory = XMLInputFactory.newInstance();
XMLOutputFactory outputFactory = XMLOutputFactory.newInstance();	    
outputFactory.setProperty("javax.xml.stream.isRepairingNamespaces", Boolean.TRUE);
		            
int count = 0;
File f = new File(xmlFileAbsoluteName);
String sourceFilePath = f.getParent();		
String tempOutputFilePrefix=sourceFilePath+"\\"+"tempNodeFile";
QName name = new QName(repeatingNode);
XMLEventReader reader;		
try {
	FileReader tempFileReader=new FileReader(xmlFileAbsoluteName);
	reader = inputFactory.createXMLEventReader(tempFileReader);
	while (reader.hasNext()) {
		XMLEvent event = reader.nextEvent();
		if(event.getEventType() == XMLStreamConstants.START_ELEMENT){
			StartElement startElement = event.asStartElement();
			if(startElement.getName().equals(name)){
			writeToFile(reader, event, tempOutputFilePrefix+ (count++) + ".xml");
			}
		}
		if (event.isEndDocument())
			break;
		}
		reader.close();
		tempFileReader.close();
			
		File renameOrigianlFile=new File(xmlFileAbsoluteName+"_processed");
		f.renameTo(renameOrigianlFile);
		int chunkSize=Integer.parseInt(numOfRecordsPerFile);
		mergeXMLFiles(sourceFilePath, splitFilesTargetDirectory+"\\"+splitFilePrefixName, chunkSize, rootNode);
		IDataUtil.put(pipelineCursor, "splitStatus", "success");
		writeToApplicationLogFile("END: Finished executing the Java code to split large XML Files. ", logFileName, project);
		} catch (FileNotFoundException e) {
			IDataUtil.put(pipelineCursor, "splitStatus", e.toString());
			writeToApplicationLogFile("ERROR: "+e.toString(), logFileName, project);
		} catch (XMLStreamException e) {
			IDataUtil.put(pipelineCursor, "splitStatus", e.toString());
			writeToApplicationLogFile("ERROR: "+e.toString(), logFileName, project);
		} catch (IOException e) {
			IDataUtil.put(pipelineCursor, "splitStatus", e.toString());
			writeToApplicationLogFile("ERROR: "+e.toString(), logFileName, project);
		}
		pipelineCursor.destroy();
	}

Shared Code:

static String logFileName="..\\IntegrationServer\\logs\\LargeFileHandling.log";
static String project="Test Calling application";

private static void writeToFile(XMLEventReader reader, XMLEvent startEvent,String filename) throws XMLStreamException, IOException {
		XMLOutputFactory outputFactory = XMLOutputFactory.newInstance();	    
		outputFactory.setProperty("javax.xml.stream.isRepairingNamespaces", Boolean.TRUE);
		XMLEventFactory eventFactory = XMLEventFactory.newInstance();
		
		StartElement element = startEvent.asStartElement();
		QName name = element.getName();
		int stack = 1;
		XMLEventWriter writer=null;
		File nodeFileName=new File(filename);
		
		FileWriter nodeFileWriter=new FileWriter(nodeFileName);
		try {
			writer = outputFactory.createXMLEventWriter(nodeFileWriter);
			writer.add(eventFactory.createStartDocument("UTF-8", "1.0"));			
			writer.add(element);			
			while (reader.hasNext()) {
			XMLEvent event = reader.nextEvent();
			if (event.isStartElement()
			        && event.asStartElement().getName().equals(name))
			    stack++;
			if (event.isEndElement()) {
			    EndElement end = event.asEndElement();
			    if (end.getName().equals(name)) {
			        stack--;
			        if (stack == 0) {
			            writer.add(event);
			            break;
			        }
			    }
			}
			writer.add(event);
			}
		} catch (Exception e) {
			writeToApplicationLogFile("ERROR: "+e.toString(), logFileName, project);
		}
		finally{			
			writer.flush();
			writer.close();
			nodeFileWriter.close();
		}
}

private static void writeToApplicationLogFile(String logData,String logFileName, String project){
		IData input = IDataFactory.create();
		IDataCursor inputCursor = input.getCursor();
		IDataUtil.put( inputCursor, "data", logData);
		IDataUtil.put( inputCursor, "logFileName", logFileName );
		IDataUtil.put( inputCursor, "projectName", project);
		inputCursor.destroy();
		IData 	output = IDataFactory.create();
		try{
			output = Service.doInvoke( "test.logging.services", "logService", input );
		}catch( Exception e){}
}

public static void mergeXMLFiles(String sourceDirectory,String destinationFileName, int chunkSize, String rootNode){
		
		writeToApplicationLogFile("START: Start executing the mergeXMLFiels static method.", logFileName, project);
		File dir = new File(sourceDirectory);
		int totalFilesCount=dir.list().length;
		int fileCount=0;
		double chunksCountDecimal=(double)totalFilesCount/(double)chunkSize;
		int chunksCount=(int) Math.ceil(chunksCountDecimal);		
		Writer outputWriter = null;
		XMLEventWriter xmlEventWriter = null;
		
		//loop to create the number of files with limited number of records.
		for (int i = 0; i < chunksCount; i++) {
			File[] rootFiles = dir.listFiles(new FilenameFilter() { 
				public boolean accept(File dir, String filename)
				     { 
				    	 return filename.endsWith(".xml"); 
				     }
				}
			);			
		int noOfFiles=rootFiles.length;		
		try {
				outputWriter = new FileWriter(destinationFileName+fileCount+".xml");
				XMLOutputFactory xmlOutFactory = XMLOutputFactory.newFactory();
				xmlEventWriter = xmlOutFactory.createXMLEventWriter(outputWriter);
				XMLEventFactory xmlEventFactory = XMLEventFactory.newFactory();
				xmlEventWriter.add(xmlEventFactory.createStartDocument("UTF-8", "1.0"));
				xmlEventWriter.add(xmlEventFactory.createStartElement("", null, rootNode));				
				XMLInputFactory xmlInFactory = XMLInputFactory.newFactory();
				for (int i1 = 0; i1 < chunkSize; i1++) {
					if(i1

Using webMethods Java service(SAX Parser) to split a huge XML file in to multiple small files.

I initially wanted to use STAX Parser which is much easy to use and efficient (From looking at some sample codes online) than the SAX parser but i finally decided to use the SAX Parser. I am currently using a webMethods 8.2.2 env with 1.6 Java so i had to use the SAX Parser to split files.

My requirement was to split a 3 GB XML file in to small XML files. This huge XML had a list of orders in them and below is a sample structure of my XML.

<?xml version="1.0" encoding="UTF-8?>

<rootNode>

       <childNode>

               <ordernumber>12354</ordernumber>



       </childNode>

       <childNode>

               <ordernumber>12355</ordernumber>



       </childNode>

       <childNode>

               <ordernumber>12356</ordernumber>



       </childNode>

</rootNode>

The result should look like this:

file1:

<?xml version="1.0" encoding="UTF-8?>

<rootNode>

       <childNode>

               <ordernumber>12354</ordernumber>



       </childNode>

       <childNode>

               <ordernumber>12355</ordernumber>



       </childNode>

</rootNode>

file2:

<?xml version="1.0" encoding="UTF-8?>

<rootNode>

       <childNode>

               <ordernumber>12356</ordernumber>



       </childNode>

</rootNode>

Java Service Code:

final IDataCursor pipelineCursor = pipeline.getCursor();
  String fileName = IDataUtil.getString( pipelineCursor, "fileName" );
  final String targetDirectotryForSplitFiles = IDataUtil.getString( pipelineCursor, "targetDirectotryForSplitFiles" );
  final String searchElement = IDataUtil.getString( pipelineCursor, "searchElement" );
  final String rootName = IDataUtil.getString( pipelineCursor, "rootName" );
  pipelineCursor.destroy();
    
  try {   
    DocumentBuilderFactory docFactory = DocumentBuilderFactory.newInstance();
    DocumentBuilder docBuilder;
    docBuilder = docFactory.newDocumentBuilder();    
    Document doc = docBuilder.parse(fileName);
    NodeList list = doc.getElementsByTagName(searchElement);
    
    final int chunkSize=list.getLength()*7;
    
    SAXParserFactory saxParserFactory = SAXParserFactory.newInstance();
    SAXParser saxParser = saxParserFactory.newSAXParser();   
    
    DefaultHandler defaultHandler=new DefaultHandler(){
    int count=0;
    int fileNameCount=0;
    ArrayList outList=new ArrayList();
    String searchTerm="close";
    public void startElement(String uri, String localName, String qName,Attributes attributes) throws SAXException {  
    if (qName.equalsIgnoreCase(searchElement)) {
     String tag = "<" + qName;
     for (int i = 0; i < attributes.getLength(); i++) {
    
      tag += " " + attributes.getLocalName(i) + "="
      + attributes.getValue(i);
     }
    
     tag += ">";
     outList.add(tag);
     searchTerm = "open";  
    }
    else{    
     if(qName.compareTo(rootName)==0){
      
     }
     else{
      outList.add("<" + qName + ">");
     }     
     }
    }  
    public void characters(char ch[], int start, int length)throws SAXException {     
     if (searchTerm.equals("open")) {
      String escapeCharPreserved=StringEscapeUtils.escapeXml(new String(ch, start, length));
      outList.add(escapeCharPreserved);
     }
    }
    public void endElement(String uri, String localName, String qName)  
    throws SAXException {
           outList.add("</" + qName + ">");
           count++;
           if (qName.equalsIgnoreCase("PCRBRecord")) {
               searchTerm = "close";
               
               if(count>chunkSize){
         try {
           writeToFile(outList,targetDirectotryForSplitFiles+"/rest_"+fileNameCount+".xml",rootName);
           fileNameCount++;
           outList.clear();
           count=0;
         } catch (IOException e) {
          IDataUtil.put(pipelineCursor, "filerror", e.toString());
         }
        }
              }
     }
    };
   
   saxParser.parse(fileName, defaultHandler);   
  } catch (ParserConfigurationException e) {
   IDataUtil.put(pipelineCursor, "result", e.toString());
  } catch (SAXException e) {
   IDataUtil.put(pipelineCursor, "result", e.toString());
  } catch (IOException e) {
   IDataUtil.put(pipelineCursor, "result", e.toString());
  }

Shared Code:

private static void writeToFile(ArrayList inputLines, String fileName, String rootName) throws IOException{
  
  ArrayList templist=new ArrayList();
  templist.add("<?xml version="1.0" encoding="UTF-8"?>");
  templist.add("<"+rootName+">");
  templist.addAll(inputLines);
  templist.add("</"+rootName+">");
  FileWriter writer = new FileWriter(fileName); 
  for(String str: templist) {
    writer.write(str);
  }
  writer.close();
 }

Imports:

import java.io.*;
import java.util.ArrayList;
import javax.xml.parsers.*;
import org.w3c.dom.Document;
import org.w3c.dom.NodeList;
import org.xml.sax.Attributes;
import org.xml.sax.SAXException;
import org.xml.sax.helpers.DefaultHandler;
import org.apache.commons.lang.StringEscapeUtils;

POP3 Email listener using Java Mail API in webMethods

webMethods provides an email port listener to retrieve emails from the inbox and execute a service with the email message contents. However for POP3, when the Integration Server picks up the message, it deletes the message from the email server. The below Java service would get emails from a POP3 Inbox and convert it to IData message structure with out deleting the messages from the Inbox.

// pipeline
IDataCursor pipelineCursor = pipeline.getCursor();
String emailHostName = IDataUtil.getString( pipelineCursor, "emailHostName" );
final String userName = IDataUtil.getString( pipelineCursor, "userName" );
final String password = IDataUtil.getString( pipelineCursor, "password" );
String port = IDataUtil.getString( pipelineCursor, "port" );
pipelineCursor.destroy();
    
try {
   
 Properties props=new Properties(); 
 props.put("mail.pop3.host" , emailHostName);
 props.put("mail.pop3.user" , userName);
 // Start SSL connection
 props.put("mail.pop3.socketFactory" , port );
 props.put("mail.pop3.socketFactory.class" , "javax.net.ssl.SSLSocketFactory" );
 props.put("mail.pop3.port" , port);
     
 Session session=Session.getInstance(props,new Authenticator() {
           @Override
    protected PasswordAuthentication getPasswordAuthentication() {
                  return new PasswordAuthentication( userName , password);
    }
  });
   
 Store store=session.getStore("pop3");
 store.connect(userName, password);
  
 Folder inbox=store.getFolder("INBOX");
 if (inbox == null) {
          System.out.println("No INBOX");
   System.exit(1);
 }
    
 //inbox.open(Folder.READ_ONLY);
 inbox.open(Folder.READ_WRITE);    
 Flags seen = new Flags(Flags.Flag.SEEN);
 FlagTerm unseenFlagTerm = new FlagTerm(seen, false);    
   
 Message[] messages = inbox.search(unseenFlagTerm);
  
 IDataUtil.put( pipelineCursor, "InboxLength", messages.length);
 ArrayList<IData> outputDoc=new ArrayList<IData>();
    
 for(int i=0;i<messages.length;i++){
         Message aMessage=messages[i];
  Object messagecontentObject= aMessage.getContent();
  
  String textMessage = null;     
  String messageUID=extractMessageID(aMessage);
  IData emailMessageResults = IDataFactory.create();
  IDataCursor emailMessageResultsCursor = emailMessageResults.getCursor();
   
  if (messagecontentObject instanceof Multipart){
             //Found Email with Attachment Retrieve the Multipart object from the message
  IDataUtil.put( emailMessageResultsCursor, "messageUID", messageUID );
  IDataUtil.put( emailMessageResultsCursor, "senderEmailAddress", new String(aMessage.getFrom()[0].toString()));
  IDataUtil.put( emailMessageResultsCursor, "subject", aMessage.getSubject());
  Multipart multipart= (Multipart) aMessage.getContent();
  ArrayList<IData> attachmentsInfoList=new ArrayList<IData>();
      
  // Loop over the parts of the email
  for (int partsCount = 0; partsCount < multipart.getCount(); partsCount++){
           BodyPart part = multipart.getBodyPart(partsCount);
       
   IData attachmentsResults=IDataFactory.create();
   IDataCursor attachmentsResultsCursor=attachmentsResults.getCursor();
       
   if (Part.ATTACHMENT.equalsIgnoreCase(part.getDisposition())) {  
   String attachmentName=null;
        if(part.getFileName()!=null){
         attachmentName = part.getFileName(); 
         
         InputStream attachmentInputStream = part.getInputStream();         
         
         //Convert the input stream from Attachment to byte[].
         ByteArrayOutputStream buffer = new ByteArrayOutputStream();         
         int nRead;
         byte[] data = new byte[16384];
         while ((nRead = attachmentInputStream.read(data, 0, data.length)) != -1) {
           buffer.write(data, 0, nRead);
           data=buffer.toByteArray();
         }
         buffer.flush();         
         
         //convert the byte[] to a base64 string.
         byte[] x=org.apache.commons.codec.binary.Base64.encodeBase64(data);
         String fileData=new String(x);
         
         String fileType=part.getContentType();         
         int fileTypeEndPos=fileType.indexOf(";");
         fileType=fileType.substring(0,fileTypeEndPos);
         
         //Populate the attachments results doc.
         IDataUtil.put( attachmentsResultsCursor, "fileName", attachmentName );
         IDataUtil.put( attachmentsResultsCursor, "fileData", fileData);
         IDataUtil.put( attachmentsResultsCursor, "fileType", fileType);
         attachmentsResultsCursor.destroy();
         attachmentsInfoList.add(attachmentsResults); 
        }
        else{
         // Unidentified attachment content type, get the part as an input byte array.
         Reader r = new InputStreamReader((InputStream) part.getContent());
         StringWriter sw = new StringWriter();
         char[] buffer = new char[1024];
         for (int n; (n = r.read(buffer)) != -1; )  
             sw.write(buffer, 0, n);
         textMessage= sw.toString();
  
         IDataUtil.put( attachmentsResultsCursor, "fileName", "No Name for attachment - Could be an attached email");
         IDataUtil.put( attachmentsResultsCursor, "fileData", textMessage);         
         attachmentsResultsCursor.destroy();
         attachmentsInfoList.add(attachmentsResults); 
        }
        
                      } else {
                       
                       //Extract inline images or other attachments                       
                       if((!part.getContentType().contains("text/html"))){
                          String attachmentName = part.getFileName();
           InputStream attachmentInputStream = part.getInputStream();         
           
           //Convert the input stream from Attachment to byte[].
           ByteArrayOutputStream buffer = new ByteArrayOutputStream();         
           int nRead;
           byte[] data = new byte[16384];
           while ((nRead = attachmentInputStream.read(data, 0, data.length)) != -1) {
             buffer.write(data, 0, nRead);
             data=buffer.toByteArray();
           }
           buffer.flush();         
           
           //convert the byte[] to a base64 string.
           byte[] x=org.apache.commons.codec.binary.Base64.encodeBase64(data);
           String fileData=new String(x);
           String fileType=part.getContentType();         
           int fileTypeEndPos=fileType.indexOf(";");
           fileType=fileType.substring(0,fileTypeEndPos);
           
           //Populate the attachments results doc.
           IDataUtil.put( attachmentsResultsCursor, "fileName", attachmentName );
           IDataUtil.put( attachmentsResultsCursor, "fileData", fileData);
           IDataUtil.put( attachmentsResultsCursor, "fileType", fileType);
           attachmentsResultsCursor.destroy();
           attachmentsInfoList.add(attachmentsResults); 
                        }
                       
                       //Handle email body for inline attachments.
                       else if( part.getContentType().contains("multipart/related")){
                        InputStreamReader is = new InputStreamReader(part.getInputStream());
          StringBuilder sb=new StringBuilder();
          BufferedReader br = new BufferedReader(is);
          String read = br.readLine();
    
          while(read != null) {
              sb.append(read);
              read =br.readLine();  
          }
          
          String unmodifiedBody=sb.toString();
          
          int htmlTagStartPos=unmodifiedBody.indexOf("<html");
          int htmlTagEndPos=unmodifiedBody.indexOf("</html>")+7;        
          String modifiedEmailBody=unmodifiedBody.substring(htmlTagStartPos,htmlTagEndPos);
          
          IDataUtil.put( emailMessageResultsCursor, "body", modifiedEmailBody);
                       }
                       // Extract email body for mails with attachment files. 
                       else{
                        textMessage = part.getContent() != null ? part.getContent().toString() : "";
                           IDataUtil.put( emailMessageResultsCursor, "body", textMessage);
                       }                                                                                 
                      }
      }
       IDataUtil.put( emailMessageResultsCursor, "emailSentDate", formatDateTime(aMessage.getSentDate()));
       IDataUtil.put( emailMessageResultsCursor, "attachmentExists", "Y" );
       IDataUtil.put( emailMessageResultsCursor, "attachmentInfo", attachmentsInfoList.toArray(new IData[attachmentsInfoList.size()]));
       outputDoc.add(emailMessageResults);      
      }
      else{
          // Email with no attachment.
         textMessage = aMessage.getContent() != null ? aMessage.getContent().toString() : "";
         IDataUtil.put( emailMessageResultsCursor, "messageUID", messageUID );
                  IDataUtil.put( emailMessageResultsCursor, "senderEmailAddress", new String(aMessage.getFrom()[0].toString()));
                  IDataUtil.put( emailMessageResultsCursor, "subject", aMessage.getSubject());
                  textMessage=aMessage.getContent().toString();
                  IDataUtil.put( emailMessageResultsCursor, "body", textMessage);
                  IDataUtil.put( emailMessageResultsCursor, "emailSentDate", formatDateTime(aMessage.getSentDate()));
                  IDataUtil.put( emailMessageResultsCursor, "attachmentExists", "N" );
   
                  IData attachmentsResults=IDataFactory.create();
                  IDataCursor attachmentsResultsCursor=attachmentsResults.getCursor();
                  IDataUtil.put( attachmentsResultsCursor, "fileName", "");
                  IDataUtil.put( attachmentsResultsCursor, "fileData", "");
                  IDataUtil.put( attachmentsResultsCursor, "fileType", "");
                  attachmentsResultsCursor.destroy();
                  
                  IDataUtil.put(emailMessageResultsCursor, "attachmentInfo", attachmentsResults);
        
                  outputDoc.add(emailMessageResults);
      }
      aMessage.setFlag(Flags.Flag.SEEN, true); // This is not supported for POP3.
    }
    inbox.close(false);
    store.close();    
    IDataUtil.put( pipelineCursor, "messageContents", outputDoc.toArray(new IData[outputDoc.size()]));    
  } catch (NoSuchProviderException e) {
   IDataUtil.put( pipelineCursor, "errrorMessage", e.toString() );
  } catch (MessagingException e) {
   IDataUtil.put( pipelineCursor, "errrorMessage", e.toString() );
  } catch (IOException e) {
   IDataUtil.put( pipelineCursor, "errrorMessage", e.toString() );
  }

Shared Code:

         private static String formatDateTime(Date date){
  
  String format="MM-dd-yyyy HH:mm:ss";  
  SimpleDateFormat sdf=new SimpleDateFormat(format);
  return sdf.format(date);
  }
  
  private static String extractMessageID(Message aMessage) throws MessagingException{
  
   // Extract message ID in proper format from the message.
   String messageUID=null;
   Enumeration headers = aMessage.getAllHeaders();                  
      while (headers.hasMoreElements()) {
          Header h = (Header) headers.nextElement();                                  
          String mID = h.getName();                
          if(mID.contains("Message-ID")){
           messageUID=h.getValue();                       
          }
      }
   
   int messageUIDLength=messageUID.length();   
   messageUID=messageUID.substring(1, messageUIDLength-1);   
   int specialCharPos=messageUID.indexOf("@");   
   messageUID=messageUID.substring(0, specialCharPos);   
   return messageUID;
  } 

consume a basic auth web service in C# and manipulate the output

This is a C# client program for a webMethods web service with basic authentication setup. This service will also loop over the output from the WS call and perform a basic extraction of the data.

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Net;

namespace reportFraudConsoleApp
{
    class Program
    {
        static void Main(string[] args)
        {
            reportfrauddataref.reportfraudVS ws = new reportfrauddataref.reportfraudVS();

            reportfrauddataref.getCaseAttachmentFilesInputDoc wsinput = new reportfrauddataref.getCaseAttachmentFilesInputDoc();
            wsinput.caseNr = "5555";

            //auth with CredentialCache class
            System.Net.CredentialCache myCred = new System.Net.CredentialCache();
            NetworkCredential networkCred = new NetworkCredential("Administrator", "wm123");
            myCred.Add(new Uri(ws.Url),"Basic",networkCred);
            ws.Credentials = myCred;

            reportfrauddataref.attachmentsData[] resultArray = ws.getCaseAttachmentFiles(wsinput);

            List fileNames=new List();
            List fileContents=new List();

            for (int i = 0; i < resultArray.Length;i++ )
            {
                // Manipulate the output from the web service by looping over the results from the ws call
                fileNames.Add(resultArray[i].fileName);
                fileContents.Add(resultArray[i].fileContent);
            }

            Console.WriteLine(fileNames.ToArray().ToString());
            Console.ReadLine();
        }
    }
}