[SNMP4J] Snmp4j traps - 1000/second

[namitha]

Hi,

I want to support for 1000 traps/second using snmp4j and have been using the
code attached in this mail.
As of now my target is to support atleast 500 traps.

I am new to this and need some help or suggestions in order to get it.

I have heard of thread pooling too for this but dont know where and how
exactly to implement here.

Any suggestions or solutions would be of great help.

Thanks in advance


-- 
NAM
-------------- next part --------------

package com.tel.Fault;

import java.io.IOException;
import java.net.UnknownHostException;

import org.apache.log4j.Logger;
import org.snmp4j.CommandResponder;
import org.snmp4j.CommandResponderEvent;
import org.snmp4j.MessageDispatcherImpl;
import org.snmp4j.Snmp;
import org.snmp4j.TransportMapping;
import org.snmp4j.mp.MPv1;
import org.snmp4j.mp.MPv2c;
import org.snmp4j.mp.MPv3;
import org.snmp4j.mp.SnmpConstants;
import org.snmp4j.security.Priv3DES;
import org.snmp4j.security.SecurityModels;
import org.snmp4j.security.SecurityProtocols;
import org.snmp4j.security.USM;
import org.snmp4j.security.UsmUser;
import org.snmp4j.smi.Address;
import org.snmp4j.smi.GenericAddress;
import org.snmp4j.smi.OID;
import org.snmp4j.smi.OctetString;
import org.snmp4j.smi.TcpAddress;
import org.snmp4j.smi.UdpAddress;
import org.snmp4j.transport.DefaultTcpTransportMapping;
import org.snmp4j.transport.DefaultUdpTransportMapping;
import org.snmp4j.util.MultiThreadedMessageDispatcher;
import org.snmp4j.util.ThreadPool;

/**
 * Classname: NMSFaultTrapListener

 * Description : This class is responsible for receiving the traps and enqueuing
 * in queue.This is also responsible for dequeuing the trap one by one from the
 * trap queue and pass it to the trap handler.
 *
 */

public class NMSFaultTrapListener  implements CommandResponder {
	
	static Logger log = Logger.getLogger(NMSFaultTrapListener.class.getName());
	private NMSFaultTrapQueue trapQueue;
	private NMSFaultTrapHandler trapHandler;
	int portNum = 0;
	private String genericAddress = null;
	/*object reference for dippatcher thread*/
	private MultiThreadedMessageDispatcher dispatcher;
	/*object reference for Snmp*/
	private Snmp snmp = null;
	/*object reference for Address*/
	private Address address = null;
	/*object reference for handling pool of threads*/
	private ThreadPool threadPool;
	/*object reference for TransportMapping*/
	private TransportMapping transport = null;
	/*declaraing and initializing thread pool size is 2 by default*/
	private int poolSize = 0; 
	/*declaraing and initializing version is version3 by default*/
	int version = SnmpConstants.version3;
	
	/*Object reference for engine ID */
	OctetString authoritativeEngineID = null;
	/*Object reference for security name*/
	OctetString securityName = null;
	/*Object reference for Authentication protocol*/
	OID authProtocol = null;
	/*Object reference for Priv protocol*/
	OID privProtocol = null;
	/*Object reference for Authentication password*/
	OctetString authPassphrase = null;
	/*Object reference for Priv password*/
	OctetString privPassphrase = null;
	
	/*address in string form*/
	String ipAddress = null;
	
	private static int count = 0;


	/**
	 * method name : NMSFaultTrapListener
	 * @param ipAddress	
	 * @param portNo
	 * @param snmpVersion
	 * @param poolSize	
	 * @return NMSFaultTrapListener
	 * @description This constructor creates instance of NMSFaultTrapQueue and initializes ipAddress,
	 * portNo, snmp version and poolSize and also starts the thread trapHandler
	 */

	public NMSFaultTrapListener  (String ipAddress, int portNo, int snmpVersion, int poolSize) {

		this.ipAddress = ipAddress;
	    this.poolSize = poolSize;
	    this.portNum = portNo;
	    this.version = snmpVersion;

	}
    	
	/**

	 * method name : setVersion
	 * @param int	
	 * @return void
	 * @description This method sets the version of the snmp
	 */
	public void setVersion(int version) {
		
		this.version = version;
	}
		
	/**

	 * method name : setVersion
	 * @param int	
	 * @return void
	 * @description This method returns the version of the snmp
	 */
	public int getVersion() {
		
		return this.version;
	}
	
	/**

	 * method name : setVersion
	 * @param int	
	 * @return void
	 * @description This method sets the port number 
	 */
	public void setPortNo(int portNo) {
		
		this.portNum = portNo;
	}
		
	/**

	 * method name : getPortNo
	 * @param void	
	 * @return int
	 * @description This method returns the port number
	 */
	public int getPortNo() {
		
		return this.portNum;
		
	}
	
	/**

	 * method name : listen()
	 * @param nil	
	 * @return void
	 * @description This method is called from receiveTrap() method in NMSFaultTrapReceiver to listen 
	 * 				to traps.This method adds usm if version is 3
	 */
	
	public void listen() throws UnknownHostException, IOException {
		try{
			
			genericAddress = this.ipAddress.trim()+"/"+new Integer(this.portNum).toString();
			address = GenericAddress.parse(genericAddress);
			/*create thread pool*/
			threadPool = ThreadPool.create("Trap Receiver ThreadPool", this.poolSize);
			/*create dispatcher*/
			dispatcher = new MultiThreadedMessageDispatcher(threadPool,new MessageDispatcherImpl());
			
			/*create transport*/
			if (address instanceof TcpAddress) {
				
				transport = new DefaultTcpTransportMapping((TcpAddress) address);
			}else {
				
				transport = new DefaultUdpTransportMapping((UdpAddress) address);
			}
			
			/*creating local Engine ID*/
			OctetString localEngineID = new OctetString(MPv3.createLocalEngineID());
			
			/*creating snmp and adding command responder*/
			snmp = new Snmp(dispatcher, transport);
			snmp.addCommandResponder(this);
			
			/*adding Message Processing Model*/
			snmp.getMessageDispatcher().addMessageProcessingModel(new MPv1());
			snmp.getMessageDispatcher().addMessageProcessingModel(new MPv2c());
			snmp.getMessageDispatcher().addMessageProcessingModel(new MPv3(localEngineID.getValue()));
			
			/*add all security protocols*/
			SecurityProtocols.getInstance().addDefaultProtocols();
			SecurityProtocols.getInstance().addPrivacyProtocol(new Priv3DES());
			
			if (version == SnmpConstants.version3) {
				USM usm = new USM(SecurityProtocols.getInstance(), localEngineID, 0);
				SecurityModels.getInstance().addSecurityModel(usm);
				
				if (authoritativeEngineID != null) {
					snmp.setLocalEngine(authoritativeEngineID.getValue(), 0, 0);
				}
				/*Add the configured user to the USM*/
				addUsmUser(snmp);
			 }
			
			snmp.listen();
			
			log.warn("NMS Listening at port "+this.portNum+ "and address "+ ipAddress);
			
		}catch(Exception e){
			
			log.error("Exception while listening to trap ", e);
		}
	}
	
	/*a private methods to add the USER*/
	private void addUsmUser(Snmp snmp) {
		try{
			
			snmp.getUSM().addUser(securityName, new UsmUser (securityName,
														 authProtocol,
														 authPassphrase,
														 privProtocol,
														 privPassphrase));
	}catch(Exception e){
		log.error("Exception while adding usm user", e);
	}
	}

	/**
	 * method name : initV3Parameter
	 * @param securityName
	 * @return void
	 * @description This method is used to initialize V3 parameters to receive 
	 * 				unauthentication traps
	 *
	 **/
	
	public void initV3Parameter(String securityName) {
		
		this.securityName = new OctetString(securityName);
		this.authProtocol=null;
		this.privProtocol=null;
		this.privPassphrase=null;
		this.authPassphrase=null;
		
		
	}

	/**
	 * method name : initV3Parameter
	 * @param securityName
	 * @param authProtocol
	 * @param authPassphrase
	 * @return void
	 * @description This method is used to initialize V3 parameters with authentication protocols 
	 * 				and password
	 **/
	
	public void initV3Parameter(String securityName,OID authProtocol,String authPassphrase) {
		
		this.securityName = new OctetString(securityName);
		this.authProtocol=authProtocol;
		this.privProtocol=null;
		this.privPassphrase=null;
		this.authPassphrase=new OctetString(authPassphrase);
			
	}
	
	/**
	 * method name : initV3Parameter
	 * @param securityName
	 * @param authProtocol
	 * @param authPassphrase
	 * @param privProtocol
	 * @param privPassphrase
	 * @return void
	 * @description This method is used to initialize V3 parameters with authentication protocols 
	 * 				and authentication password and also with privacy protocols and privacy password
	 **/
	
	public void initV3Parameter(String securityName,OID authProtocol,String authPassphrase,OID privProtocol,String privPassphrase) {
		
		this.securityName = new OctetString(securityName);
		this.authProtocol=authProtocol;
		this.privProtocol=privProtocol;
		this.privPassphrase=new OctetString(privPassphrase);
		this.authPassphrase=new OctetString(authPassphrase);
		
		
	}
	
	/**
	 * method name : stop
	 * @param void
	 * @return void
	 * @description This method removes the command responder and closes listener for snmp
	 **/
	public void stop() throws IOException {
	
		snmp.removeCommandResponder(this);
		snmp.close();
		threadPool.cancel();		
		
	}
	
	/**
	 * method name : processPdu
	 * @param event
	 * @return void
	 * @description This method is called automatically when traps are received in order to process the
	 * 				trap. This method adds the received traps to queue and notify the observers of the 
	 * 				trap event
	 **/
	
	@Override
	public synchronized void processPdu(CommandResponderEvent event) {
		try{
			log.error("trap received for"+count++ +"times");
			// TODO Auto-generated method stub
			log.warn("Trap Received"+event);
			trapQueue.enqueue(event); // Add the trap in the TrapQueue
			
		}catch(Exception e){
			
				log.error("Exception while processing Trap PDU", e);
		}
		
	}
	
	public static void main(String args[]){
		NMSFaultTrapListener list = new NMSFaultTrapListener("10.50.34.36",162,1,2);
		try {
			list.listen();
		} catch (UnknownHostException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
	}
}