[SNMP4J] Extending Agent

[Eugene R. Snider]

Here's some code fragments you might find useful for your UDP stuff. The 
MultiThreadedUDPReciever class will hang out on a UDP port of your 
specification and process incoming messages in multiple threads. When a 
message is recieved the UDP stream is parsed by the (not defined) 
UDPEvent constructor. This turns the UDP data stream into an event class.

Other classes may listen for incoming events by implmenting the 
UDPMonitor interface defined below. Classes register for Monitor events 
by type (also not included, I refer to an external UDPMonitorConstants 
class which is implementation specific ) using the static method 
addListener.

When parsing is complete the registered listeners are notified by via 
the UDPMonitor interface handleEvent callback based on their registered 
event types. The event handlers (which are probably event specific) then 
take the implementation specific event class and may manipulate it in 
any way they desire. The event handlers are where table management and 
other agent housekeeping tasks should be performed, additionally 
implmentation specific traps related to the event may be thrown from the 
event handlers. These are normally co-located in the mib class 
implementation so that each mib listens for the events which are of 
interest to it and manages its interface. An example would be the 
ifMib.java class which updates the ifTable and an enterprise or private 
mib handler which manages the enterprise specific objects.

The event notifier expects two defined event types, EVENT_TYPE_UNKNOWN 
and EVENT_TYPE_ALL

Gene

--- Begin UDPMonitorConstants

package com.mrcompiler.snmp.server.agent;

public class UDPMonitorConstants {
    public static enum MonitorEventType {
        EVENT_TYPE_NONE,
        ALL_EVENTS,       
        UDP_EVENT_ONE,
        UDP_EVENT_TWO,
        UNKNOWN_EVENT_TYPE
    }
}


-- Begin MutliThreadedUDPListener class

package com.mrcompiler.snmp.server.agent;

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

import org.snmp4j.TransportMapping;
import org.snmp4j.log.LogAdapter;
import org.snmp4j.log.LogFactory;
import org.snmp4j.smi.Address;
import org.snmp4j.smi.GenericAddress;
import org.snmp4j.smi.UdpAddress;
import org.snmp4j.transport.DefaultUdpTransportMapping;
import org.snmp4j.transport.TransportListener;
import org.snmp4j.transport.UdpTransportMapping;
import org.snmp4j.util.ThreadPool;

import 
com.mrcompiler.snmp.server.agent.UDPMonitorConstants.MonitorEventType;
import com.mrcompiler.snmp.server.views.PlanMonitorView;

public class MultiThreadedUDPReceiver implements TransportListener {
    private int[] types;

    private UDPMonitorEventListener[] listeners;

    private int level;

    private static final LogAdapter logger = LogFactory
            .getLogger(MultiThreadedUDPReceiver.class);

    public MultiThreadedUDPReceiver() {
    }

    public void addListener(MonitorEventType eventType,
            UDPMonitorEventListener listener) {
        if (types == null)
            types = new int[4];
        if (listeners == null)
            listeners = new UDPMonitorEventListener[4];
        int length = types.length, index = length - 1;
        while (index >= 0) {
            if (types[index] != 0)
                break;
            --index;
        }
        index++;
        if (index == length) {
            int[] newTypes = new int[length + 4];
            System.arraycopy(types, 0, newTypes, 0, length);
            types = newTypes;
            UDPMonitorEventListener[] newListeners = new 
UDPMonitorEventListener[length + 4];
            System.arraycopy(listeners, 0, newListeners, 0, length);
            listeners = newListeners;
        }
        types[index] = eventType.ordinal();
        listeners[index] = listener;

    }

    void remove(int index) {
        if (level == 0) {
            int end = types.length - 1;
            System.arraycopy(types, index + 1, types, index, end - index);
            System.arraycopy(listeners, index + 1, listeners, index, end
                    - index);
            index = end;
        } else {
            if (level > 0)
                level = -level;
        }
        types[index] = 0;
        listeners[index] = null;
    }

    public void removeListener(int eventType, Listener listener) {
        if (types == null)
            return;
        for (int i = 0; i < types.length; i++) {
            if (types[i] == eventType && listeners[i] == listener) {
                remove(i);
                return;
            }
        }
    }

    synchronized void postEvent(MonitorEventType eventType,
            UDPMonitorEvent event) {
        if (types == null)
            return;
        level += level >= 0 ? 1 : -1;
        try {
            for (int i = 0; i < types.length; i++) {
                if ((event.type == MonitorEventType.UNKNOWN_EVENT_TYPE 
&& types[i] == MonitorEventType.ALL_EVENTS
                        .ordinal())
                        || (types[i] == 
MonitorEventType.ALL_EVENTS.ordinal() || types[i] == event.type
                                .ordinal())) {
                    UDPMonitorEventListener listener = listeners[i];
                    if (listener != null)
                        listener.handleEvent(eventType, event);
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            boolean compact = level < 0;
            level -= level >= 0 ? 1 : -1;
            if (compact && level == 0) {
                int index = 0;
                for (int i = 0; i < types.length; i++) {
                    if (types[i] != 0) {
                        types[index] = types[i];
                        listeners[index] = listeners[i];
                        index++;
                    }
                }
                for (int i = index; i < types.length; i++) {
                    types[i] = 0;
                    listeners[i] = null;
                }
            }
        }
    }

    public class UDPMessageProcessingTask implements Runnable {
        private TransportMapping sourceTransport;

        private Address incomingAddress;

        private ByteBuffer wholeMessage;

        public UDPMessageProcessingTask() {
            return;
        }

        public UDPMessageProcessingTask(TransportMapping sourceTransport,
                Address incomingAddress, ByteBuffer wholeMessage) {
            this.sourceTransport = sourceTransport;
            this.incomingAddress = incomingAddress;
            this.wholeMessage = wholeMessage;
        }

        public void run() {
            // TODO Auto-generated method stub
            String eventData = new String(wholeMessage.array());
            logger.info("Processing UDP Message " + eventData);
            UDPMonitorEvent eventInfo = new UDPMonitorEvent(eventData);

            // if( eventInfo.type != MonitorEventType.UNKNOWN_EVENT_TYPE ) {
            postEvent(eventInfo.type, eventInfo);
            // }
        }

        public Address getIncomingAddress() {
            return incomingAddress;
        }

        public void setIncomingAddress(Address incomingAddress) {
            this.incomingAddress = incomingAddress;
        }

        public TransportMapping getSourceTransport() {
            return sourceTransport;
        }

        public void setSourceTransport(TransportMapping sourceTransport) {
            this.sourceTransport = sourceTransport;
        }

        public ByteBuffer getWholeMessage() {
            return wholeMessage;
        }

        public void setWholeMessage(ByteBuffer wholeMessage) {
            this.wholeMessage = wholeMessage;
        }
    }

    private Address listenAddress;

    private ThreadPool threadPool;

    private UdpTransportMapping transport;

    private void init() throws UnknownHostException, IOException {
        if (threadPool == null) {
            threadPool = ThreadPool.create("UDPStreams", 2);
        }

        if (listenAddress == null) {
            listenAddress = GenericAddress.parse(System.getProperty(
                    "snmpAgent.udpListenAddress", "udp:0.0.0.0/1234"));
        }
        if (transport == null) {
            transport = new DefaultUdpTransportMapping(
                    (UdpAddress) listenAddress);
            transport.addTransportListener(this);
        }
    }

    public void run() {
        try {
            init();
            transport.listen(); // start the listener thread
        } catch (Exception ex) {
            ex.printStackTrace();
        }
    }

    public void processMessage(TransportMapping sourceTransport,
            Address incomingAddress, ByteBuffer wholeMessage) {
        UDPMessageProcessingTask task = new UDPMessageProcessingTask(
                sourceTransport, incomingAddress, wholeMessage);
        threadPool.execute(task);
    }

    public Address getListenAddress() {
        return listenAddress;
    }

    public void setListenAddress(Address listenAddress) {
        this.listenAddress = listenAddress;
    }

    public void stop() throws Throwable {
        super.finalize();
    }

}

---- End of MultiThreadedUDPListener.class

-- Begin

package com.mrcompiler.snmp.server.agent;

import java.util.EventListener;

import 
com.mrcompiler.snmp.server.agent.UDPMonitorConstants.MonitorEventType;

public interface UDPMonitor extends EventListener {

    void handleEvent (MonitorEventType eventType, UDPMonitorEvent event);

}