Polling Notifications

The implementation of a polling notification is similar to the implementation of an adapter service. Each implementation includes a Java class extending an ADK base class, and a namespace node in which design-time configuration data is stored. In the Java class, the metadata model for polling notifications is nearly identical to that of adapter services. The parameters to configure a polling notification are built from the polling notifications' metadata.

The primary difference between adapter services and polling notifications is the runtime behavior of polling notifications. Polling notifications cannot be directly invoked from a flow service or from Designer. Instead, Integration Server automatically invokes a polling notification in a fixed time interval. When a polling notification determines that a specified event has occurred in the adapter resource, it produces a document describing the event. These documents are automatically published to Integration Server (or IBM webMethods Broker) as they are generated by the notification. The processing of the published document is based on triggers that are configured to invoke flow services when the given document type is published.

Create a polling notification service by extending com.wm.adk.notification.WmPollingNotification base class . You must override the following base class methods in your WmPollingNotification implementation class:

package com.wm.MyAdapter.notifications;

import com.wm.adk.cci.record.WmRecord;
import com.wm.adk.cci.record.WmRecordFactory;
import com.wm.adk.connection.WmManagedConnection;
import com.wm.adk.error.AdapterException;
import com.wm.adk.metadata.ResourceDomainValues;
import com.wm.adk.metadata.WmAdapterAccess;
import com.wm.adk.metadata.WmTemplateDescriptor;
import com.wm.adk.notification.WmPollingNotification;

import java.io.File;
import java.util.ArrayList;
import java.util.Locale;
import javax.resource.ResourceException;

import com.wm.MyAdapter.MyAdapter;

public class SimpleNotification extends WmPollingNotification
{
	public static final String NOTIFICATION_SETUP_GROUP = "SimpleNotification";

	public static final String DIRECTORY_PARM = "directory";
	public static final String CHECK_ADDED_PARM = "checkAdded";
	public static final String CHECK_DELETED_PARM = "checkDeleted";
	public static final String SIG_FIELD_NAMES_PARM = "fieldNames";
	public static final String SIG_FIELD_TYPES_PARM = "fieldTypes";
	public static final String SIG_PARM = "signature";

	public static final String DIRECTORIES_RD =
		"SimpleNotification.directories.rd";
	public static final String FIELD_NAMES_RD =
		"SimpleNotification.fieldNames.rd";
	public static final String FIELD_TYPES_RD =
		"SimpleNotification.fieldTypes.rd";

	private String _directory;
	private boolean _checkAdded;
	private boolean _checkDeleted;
	private String[] _fieldNames;
	private String[] _fieldTypes;

	public void setDirectory(String val){_directory = val;}
	public void setCheckAdded(boolean val){_checkAdded = val;}
	public void setCheckDeleted(boolean val){_checkDeleted = val;}
	public void setFieldNames(String[] val){_fieldNames = val;}
	public void setFieldTypes(String[] val){_fieldTypes = val;}
	public void setSignature(String[] val){}

	private ArrayList _fileList = new ArrayList();

	public SimpleNotification(){}

	public void fillWmTemplateDescriptor(WmTemplateDescriptor descriptor, Locale l)
		throws ResourceException
	{
		descriptor.createGroup(NOTIFICATION_SETUP_GROUP,
			new String[]{DIRECTORY_PARM, CHECK_ADDED_PARM, CHECK_DELETED_PARM,
				SIG_FIELD_NAMES_PARM, SIG_FIELD_TYPES_PARM, SIG_PARM});
		descriptor.createFieldMap(
			new String[]{SIG_FIELD_NAMES_PARM, SIG_FIELD_TYPES_PARM, SIG_PARM},
			false);

		descriptor.setHidden(SIG_FIELD_NAMES_PARM);
		descriptor.setHidden(SIG_FIELD_TYPES_PARM);
		descriptor.setHidden(SIG_PARM);
		descriptor.createTuple(
			new String[]{SIG_FIELD_NAMES_PARM, SIG_FIELD_TYPES_PARM});

		descriptor.setResourceDomain(DIRECTORY_PARM, DIRECTORIES_RD, null);
		descriptor.setResourceDomain(SIG_FIELD_NAMES_PARM, FIELD_NAMES_RD, null);
		descriptor.setResourceDomain(SIG_FIELD_TYPES_PARM, FIELD_TYPES_RD, null);
		descriptor.setResourceDomain(SIG_PARM, WmTemplateDescriptor.OUTPUT_FIELD_NAMES,
			new String[]{SIG_FIELD_NAMES_PARM, SIG_FIELD_TYPES_PARM});
		descriptor.setDescriptions(
			MyAdapter.getInstance().getAdapterResourceBundleManager(),l);
	}

	public void runNotification() throws ResourceException
	{
		File thisDir = new File(_directory);
		File [] newList = thisDir.listFiles();
		ArrayList scratchCopy = new ArrayList(this._fileList);

		for (int nlIndex = 0;nlIndex < newList.length;nlIndex++)
		{
			String name = newList[nlIndex].getName();
			if(newList[nlIndex].isFile())
			{
				if(scratchCopy.contains(name))
					{
						scratchCopy.remove(name);
					}
				else
				{
					this._fileList.add(name);
					if(this._checkAdded)
					{
						this.doNotify(createNotice(name,_directory,true,false));
					}
				}
			}
			else
			{
				scratchCopy.remove(name);
			}

		}
		// now anything left in the scratch copy is missing from the directory

		String[] deadList = new String[scratchCopy.size()];
		scratchCopy.toArray(deadList);
		for(int dlIndex = 0; dlIndex < deadList.length;dlIndex++)
		{
			 this._fileList.remove(deadList[dlIndex]);
			 if(this._checkDeleted)
			 {
				 this.doNotify(createNotice(deadList[dlIndex], _directory,false,true));
			 }
		}

	}


	public Boolean adapterCheckValue(WmManagedConnection connection,
		String resourceDomainName, String[][] values, String testValue)
		throws AdapterException
	{

		boolean result = true;

		if(resourceDomainName.equals(DIRECTORIES_RD))
		{
			File testDir = new File(testValue);
			if (!testDir.exists())
			{
				result = false;
			}
			else if(!testDir.isDirectory())
			{
				result = false;
			}
		}

		return new Boolean(result);
	}

	public ResourceDomainValues[] adapterResourceDomainLookup(
		WmManagedConnection connection, String resourceDomainName,
		String[][] values) 	throws AdapterException
	{
		ResourceDomainValues[] results = null;

		if (resourceDomainName.equals(FIELD_NAMES_RD) || 
			resourceDomainName.equals(FIELD_TYPES_RD))
		{
			ResourceDomainValues names =
				new ResourceDomainValues(FIELD_NAMES_RD,new String[] {
				"FileName", "Path","isAdded","isDeleted"});

			ResourceDomainValues types =
				new ResourceDomainValues(FIELD_TYPES_RD,new String[] {
				"java.lang.String", "java.lang.String",
				"java.lang.Boolean","java.lang.Boolean"});
			results = new ResourceDomainValues[] {names,types};
		}

		return results;
	}

	public void registerResourceDomain(	WmManagedConnection connection,
		WmAdapterAccess access) throws AdapterException
	{
		access.addResourceDomainLookup(this.getClass().getName(),
			FIELD_NAMES_RD,connection);
		access.addResourceDomainLookup(this.getClass().getName(),
			FIELD_TYPES_RD,connection);

		ResourceDomainValues rd = new ResourceDomainValues(DIRECTORIES_RD,
			new String[] {""});
		rd.setComplete(false);
		rd.setCanValidate(true);
		access.addResourceDomain(rd);
		access.addCheckValue(DIRECTORIES_RD, connection);
	}

	private WmRecord createNotice(String file, String dir, boolean isAdded, 
		boolean isDeleted)
	{
		WmRecord notice =
			WmRecordFactory.getFactory().createWmRecord("notUsed");
		notice.put("FileName",file);
		notice.put("Path",dir);
		notice.put("isAdded",new Boolean(isAdded));
		notice.put("isDeleted", new Boolean(isDeleted));

		return notice;
	}

}

The next step for implementing a polling notification is to create the metadata constructs that the users of the adapter use for entering data when they create polling notification nodes. To do this, you perform the following:

• Create metadata parameters appropriate for the function of the polling notification. Each parameter has:
  • A variable to hold the configured values.
  • A String constant containing the name of the parameter.
  • An accessor method.
  • A set of resource bundle entries with a localizable parameter name and description as shown in Updating the Resource Bundle.

  For more information on metadata parameters, see Metadata Model for Connection.

• Describe presentation for those metadata parameters.
• Set the data entry rules for those metadata parameters.

The example implementation includes three metadata parameters that the users of the adapter use to create polling notification nodes. The following table describes the purpose of each of these parameters for data entry:

After you implement the configuration logic for the polling notification, you implement the logic that defines the signature of the polling notification node. It is used to create a document type node that enables triggers to identify notification data in the node.

To define an output signature, you create additional metadata parameters as follows:

With few exceptions, the document properties for an adapter's polling and listener notifications are managed and manipulated the same way as they are for an adapter service's signature. For the three template based features (signature wrapping, override connection name, and pass full pipeline), only the pass full pipeline feature applies to notification documents, and then, only for synchronous notifications. When the pass full pipeline option is enabled for a synchronous listener notification, then the notification can pass fields to the invoked service that are not defined in its request document, and receive fields from that service that are not defined in the notification's reply document. Template-based signature manipulation features have no other effect on notification documents.

Document field properties are managed in exactly the same way as the signature field properties as described earlier. In the case of notifications, the setSignatureProperties method is called setDocumentProperties method. For asynchronous notifications, only one PipelineRecordProperties argument exists.

The resource domain implementation for the signature parameters in this example is straightforward. The signature in this case is static, but it is implemented as a lookup, to facilitate maintenance of the class.

The final task for implementing a polling notification is to add the runNotification method and any callback methods. This example implements some very basic logic as previously described. It relies on the fact that the object instance is reused between runNotification calls. This may not be a good technique if the runNotification call runs for a long time or if overlapping calls occur. A more robust model would probably use a persistent store instead of an instance variable to track the current directory snapshot.

Update the resource bundle with display names, and descriptions to make the polling notification more usable, as follows:

package com.wm.MyAdapter;
..
..
import com.wm.MyAdapter.notifications.SimpleNotification;

public class MyAdapterResource extends ListResourceBundle implements MyAdapterConstants{
   ..
   .. 
 static final Object[][] _contents = {	
   ..
   .. 	
		//Polling Notifications
		,{SimpleNotification.class.getName() + ADKGLOBAL.RESOURCEBUNDLEKEY_DISPLAYNAME,
			"Simple Polling Notification"}
		,{SimpleNotification.class.getName() + ADKGLOBAL.RESOURCEBUNDLEKEY_DESCRIPTION,
			"Looks for file updates to a specified directory"}
		,{SimpleNotification.NOTIFICATION_SETUP_GROUP + ADKGLOBAL.RESOURCEBUNDLEKEY_GROUP, 
			"Simple Notification Settings"}
		,{SimpleNotification.DIRECTORY_PARM + ADKGLOBAL.RESOURCEBUNDLEKEY_DISPLAYNAME,
			"Directory Path"}
		,{SimpleNotification.DIRECTORY_PARM + ADKGLOBAL.RESOURCEBUNDLEKEY_DESCRIPTION,
			"Directory to monitor"}
		,{SimpleNotification.CHECK_ADDED_PARM + ADKGLOBAL.RESOURCEBUNDLEKEY_DISPLAYNAME,
			"Notify on Add"}
		,{SimpleNotification.CHECK_ADDED_PARM + ADKGLOBAL.RESOURCEBUNDLEKEY_DESCRIPTION,
			"Check if notification must be generated when file added"}
		,{SimpleNotification.CHECK_DELETED_PARM + ADKGLOBAL.RESOURCEBUNDLEKEY_DISPLAYNAME,
			"Notify on Delete"}
		,{SimpleNotification.CHECK_DELETED_PARM + ADKGLOBAL.RESOURCEBUNDLEKEY_DESCRIPTION,
			"Check if notification must be generated when file deleted"}

  }
  protected Object[][] getContents() {
		 // TODO Auto-generated method stub
		 return _contents;
  }
}

You must register each polling notification class in the WmAdapter implementation class. You do this by passing the class name to the AdapterTypeInfo.addNotificationType method in the WmAdapter.fillAdapterTypeInfo method in WmAdapter implementation class. In the example, the polling notification SimpleNotification class is registered in the MyAdapter adapter implementation class:

For example:

package com.wm.MyAdapter;
..
..
import com.wm.MyAdapter.notifications.*;  
..
..  
public class MyAdapter extends WmAdapter implements MyAdapterConstants {
..
.. 
	public void fillAdapterTypeInfo(AdapterTypeInfo info, Locale locale)  
	{
    ..
    ..
    info.addNotificationType(SimpleNotification.class.getName());
 }
}

When a callback is coordinated across the cluster, that callback is only executed on one instance of the polling notification in the cluster. For example, if the enableCallBack is coordinated, the first polling notification in the cluster that is enabled, executes the enableCallBack. When subsequent instances of that notification are enabled, the enableCallBack call is suppressed. If the callbacks were not coordinated, each instance executes the enableCallBack when the node instance is enabled.

The purpose of callback coordination is to prevent redundant updates to the backend associated with starting or stopping a notification. For example, when the disableCallBack is called on a polling notification for the Adapter for JDBC, a database trigger that gathers information for the notification is removed. Without coordination, all instances of that notification would be effectively disabled as soon as the first instance is disabled. With coordination, the disableCallBack is not executed until the last instance of that notification in the cluster is disabled.

From a design standpoint, it is important to segregate management of resources on the backend in separate callbacks from management of resources that are local to the notification instance. Callback coordination is configured so that related pairs of callbacks are either coordinated, or not. The following pairs of callbacks may be coordinated:

For information on how to configure callback coordination, see Configuration Settings.

When a polling notification is started, it executes polls in an interval according to its schedule configuration. When polling is coordinated across a cluster, a poll is executed on only one of the active instances at each scheduled interval. Which instance executes depends on the configuration and timing.

Integration Server supports a coordination mode configuration setting for each polling notification node. This configuration is normally set from Integration Server Administrator on the same page where the schedule is set. The following coordination mode values are supported:

The coordination information for clustered polling notifications is stored in the cluster’s shared cache.

All coordination of clustered polling notifications is done through an entry specific to the node in the cluster's shared cache. When the first instance of a notification (with coordination enabled) is introduced into a cluster, a new shared cache session is created and populated with the name of the server hosting the node, and the state of the node. As that node is copied to other servers in the cluster, each instance is registered in the same shared cache session. The states of these respective instances are used to determine when coordinated callbacks must be executed.

When the first instance of a clustered polling notification is started, that instance is marked as "primary", its schedule and coordination settings are recorded, and the time calculated for the next poll are all recorded in the shared cache session. Being "primary" means that first instance executes the first poll (barring failures). When the "primary" schedules the next poll, it releases the "primary" status if coordination is distributed, or retains it if coordination is configured for standby mode.

When another instance of a clustered polling notification starts, that instance first detects that a previous instance is already started. Since this instance is not the first, it overwrites its own cluster and schedule settings with those recorded in the shared cache. This instance then schedules itself to "wake up" at the next scheduled poll time. When it wakes up, if another instance is marked as "primary", the notification instance will verify that the indicated instance is still active, then reschedule itself to wake up periodically until it detects that the poll was completed within the configured time limit. It then reschedules itself to wake up at the next scheduled poll and repeats the process. If the polling time arrives and no instance is marked as "primary", or it detects that the "primary" instance is no longer functioning, then the local instance assumes the "primary" role and executes the poll as described above. Since all coordination is based on timestamps recorded in the shared cache, it is very important for server clocks to be synchronized.

Cluster coordination is controlled by a number of configuration settings to control behavior and tune failure detection using timeouts.

<?xml version="1.0"?>  
<clusterProps>  
  <pollingNotifications>  
    <callbackScheme>1</callbackScheme>  
    <runtimeModeLimit>distribute</runtimeModeLimit>  
    <template className="com.wm.adapter.wmarttest.notification.LatchedPollingNotification">  
       <callbackScheme>1</callbackScheme>  
       <runtimeModeLimit>standby</runtimeModeLimit>  
    </template>  
  </pollingNotifications>  
  <listenerNotifications>  
    <callbackScheme>1</callbackScheme>  
  </listenerNotifications>  
  <listeners>  
    <runtimeModeLimit>standby</runtimeModeLimit>  
  </listeners>  
</clusterProps>