Added Threads Pool and Dispatcher

SNMP_Pipeline/SnmpReciever/src/main/java/com/example/SnmpReciever/SnmpReceiverApplication.java

 package com.example.SnmpReciever;
 
-import org.snmp4j.PDU;
-import org.snmp4j.smi.*;
+import com.example.SnmpReciever.component.SnmpListener;
 import org.springframework.boot.CommandLineRunner;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 
-import java.io.IOException;
-
 @SpringBootApplication
 public class SnmpReceiverApplication implements CommandLineRunner {
 	private final SnmpListener trapReceiver;
@@ -21,6 +18,6 @@ public class SnmpReceiverApplication implements CommandLineRunner {
 
 	@Override
 	public void run(String... args) throws Exception {
-		trapReceiver.startTrapListener();
+		trapReceiver.listen();
 	}
 }
\ No newline at end of file

SNMP_Pipeline/SnmpReciever/src/main/java/com/example/SnmpReciever/SnmpListener.java → SNMP_Pipeline/SnmpReciever/src/main/java/com/example/SnmpReciever/component/SnmpListener.java

-package com.example.SnmpReciever;
-import com.example.SnmpReciever.enitites.TrapData;
-//import com.example.SnmpReciever.services.KafkaService;
-import com.fasterxml.jackson.core.JsonProcessingException;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import lombok.AllArgsConstructor;
-import org.snmp4j.*;
-import org.snmp4j.smi.UdpAddress;
-import org.snmp4j.transport.DefaultUdpTransportMapping;
-import org.springframework.kafka.core.KafkaTemplate;
-import org.springframework.stereotype.Component;
-
-import java.io.IOException;
-
-/*@RestController
-@RequestMapping("/api/product")*/
-@Component
-@AllArgsConstructor
-public class SnmpListener implements CommandResponder {
-
-    private final KafkaTemplate<String, String> kafkaTemplate;
-    private final ObjectMapper objectMapper;
-
-
-    @Override
-    public synchronized void processPdu(CommandResponderEvent event) {
-        System.out.println("Received PDU...");
-        PDU x = event.getPDU();
-//        System.out.println(x);
-//        System.out.println(x.getType());
-//
-        PDUv1 pdu = (PDUv1) x;
-        TrapData trapData = new TrapData(pdu);
-        if (pdu != null) {
-            try {
-                String pduJson = objectMapper.writeValueAsString(trapData);
-//                System.out.println(pduJson);
-                kafkaTemplate.send("TRAP", pduJson);
-            } catch (JsonProcessingException e) {
-                System.out.println(e);
-                throw new RuntimeException(e);
-            }
-        }
-    }
-    public void startTrapListener() {
-        try {
-            TransportMapping<?> transport = new DefaultUdpTransportMapping(new UdpAddress("192.168.26.46/1625"));
-            System.out.println("Listening to SNMP Trap");
-            Snmp snmp = new Snmp(transport);
-            snmp.addCommandResponder(this);
-            transport.listen();
-        } catch (IOException e) {
-            e.printStackTrace();
-        }
-    }
-
-}
+package com.example.SnmpReciever.component;
+import com.example.SnmpReciever.enitites.TrapData;
+//import com.example.SnmpReciever.services.KafkaService;
+import com.example.SnmpReciever.services.KafkaService;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.AllArgsConstructor;
+import org.snmp4j.*;
+import org.snmp4j.mp.MPv1;
+import org.snmp4j.mp.MPv2c;
+import org.snmp4j.security.Priv3DES;
+import org.snmp4j.security.SecurityProtocols;
+import org.snmp4j.smi.OctetString;
+import org.snmp4j.smi.TcpAddress;
+import org.snmp4j.smi.TransportIpAddress;
+import org.snmp4j.smi.UdpAddress;
+import org.snmp4j.transport.AbstractTransportMapping;
+import org.snmp4j.transport.DefaultTcpTransportMapping;
+import org.snmp4j.transport.DefaultUdpTransportMapping;
+import org.snmp4j.util.MultiThreadedMessageDispatcher;
+import org.snmp4j.util.ThreadPool;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.stereotype.Component;
+
+import java.io.IOException;
+
+/*@RestController
+@RequestMapping("/api/product")*/
+@Component
+public class SnmpListener implements CommandResponder {
+
+    @Autowired
+    private KafkaService kafkaService;
+    private ObjectMapper objectMapper = new ObjectMapper();
+
+    @Value("${spring.listener.address}")
+    private String addressString;
+
+    @Value("${spring.listener.threads}")
+    private Integer threadsNum;
+
+    @Value("${spring.listener.community}")
+    private String community;
+
+
+    @Override
+    public synchronized void processPdu(CommandResponderEvent event) {
+        System.out.println("Received PDU...");
+        PDU x = event.getPDU();
+        PDUv1 pdu = (PDUv1) x;
+        TrapData trapData = new TrapData(pdu);
+        if (pdu != null) {
+            try {
+                String pduJson = objectMapper.writeValueAsString(trapData);
+                kafkaService.send("TRAP", pduJson);
+            } catch (JsonProcessingException e) {
+                System.out.println(e);
+                throw new RuntimeException(e);
+            }
+        }
+    }
+    public void startTrapListener() {
+        try {
+            TransportMapping<?> transport = new DefaultUdpTransportMapping(new UdpAddress(addressString));
+            System.out.println("Listening to SNMP Trap");
+            Snmp snmp = new Snmp(transport);
+            snmp.addCommandResponder(this);
+            transport.listen();
+        } catch (IOException e) {
+            e.printStackTrace();
+        }
+    }
+    public synchronized void listen()
+            throws IOException {
+        AbstractTransportMapping transport;
+        TransportIpAddress address = new UdpAddress(addressString);
+        if (address instanceof TcpAddress) {
+            transport = new DefaultTcpTransportMapping((TcpAddress) address);
+        } else {
+            transport = new DefaultUdpTransportMapping((UdpAddress) address);
+        }
+
+        ThreadPool threadPool = ThreadPool.create("DispatcherPool", threadsNum);
+        MessageDispatcher mDispatcher = new MultiThreadedMessageDispatcher(
+                threadPool, new MessageDispatcherImpl());
+
+        // add message processing models
+        mDispatcher.addMessageProcessingModel(new MPv1());
+        mDispatcher.addMessageProcessingModel(new MPv2c());
+
+        // add all security protocols
+        SecurityProtocols.getInstance().addDefaultProtocols();
+        SecurityProtocols.getInstance().addPrivacyProtocol(new Priv3DES());
+
+        // Create Target
+        CommunityTarget target = new CommunityTarget();
+        target.setCommunity(new OctetString(community));
+
+        Snmp snmp = new Snmp(mDispatcher, transport);
+        snmp.addCommandResponder(this);
+
+        transport.listen();
+        System.out.println("Listening on " + address);
+        try {
+            this.wait();
+        } catch (InterruptedException ex) {
+            Thread.currentThread().interrupt();
+        }
+    }
+
+}

SNMP_Pipeline/SnmpReciever/src/main/java/com/example/SnmpReciever/services/KafkaService.java

-//package com.example.SnmpReciever.services;
-//
-//import com.fasterxml.jackson.core.JsonProcessingException;
-//import com.fasterxml.jackson.databind.ObjectMapper;
-//import lombok.AllArgsConstructor;
-//import org.springframework.kafka.core.KafkaTemplate;
-//import org.springframework.stereotype.Service;
-//
-//@Service
-//@AllArgsConstructor
-//public class KafkaService {
-//    private final KafkaTemplate<String, String> kafkaTemplate;
-//
-//    public void send(String topic,String message){
-//        kafkaTemplate.send(topic,message);
-//    }
-//}
+package com.example.SnmpReciever.services;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.AllArgsConstructor;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.stereotype.Service;
+
+@Service
+@AllArgsConstructor
+public class KafkaService {
+    private final KafkaTemplate<String, String> kafkaTemplate;
+
+    public void send(String topic,String message){
+        kafkaTemplate.send(topic,message);
+    }
+}

SNMP_Pipeline/SnmpReciever/src/main/resources/application.properties

@@ -2,4 +2,6 @@ server.port = 0
 spring.kafka.bootstrap-servers=http://172.29.3.220:9092
 spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
 spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
-
+spring.listener.address=192.168.26.46/1625
+spring.listener.threads=20
+spring.listener.community=public

SNMP_Pipeline/SnmpReciever/target/classes/application.properties

@@ -2,4 +2,6 @@ server.port = 0
 spring.kafka.bootstrap-servers=http://172.29.3.220:9092
 spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
 spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
-
+spring.listener.address=192.168.26.46/1625
+spring.listener.threads=20
+spring.listener.community=public

SNMP_Pipeline/kafkaConsumerSNMP/src/main/java/com/example/kafkaConsumerSNMP/KafkaConsumerSnmpApplication.java

@@ -25,19 +25,18 @@ public class KafkaConsumerSnmpApplication {
 
 	@KafkaListener(topics = "TRAP")
 	public void handleKafkaMessage(String pduJson) {
-		String json = pduJson; // Replace with your JSON string
+		String json = pduJson;// Replace with your JSON string
 		num++;
 		try {
 			ObjectMapper objectMapper = new ObjectMapper();
 			TrapData t = objectMapper.readValue(json,TrapData.class);
 			//Some Processing: Add this later
 			EnrichedTrap t2 = new EnrichedTrap(t);
-			System.out.println("Not sent yet");
+			//System.out.println("Not sent yet");
 			sender.send(t2);
-			System.out.println("Sent now!");
+			//System.out.println("Sent now!");
 			rethinkDBService.saveKafkaMessageToRethink(t2);
-			if (num == 1)
-				System.out.println(rethinkDBService.getData());
+			System.out.println(num);
 		} catch (Exception e) {
 			System.out.println(e);
 			e.printStackTrace();

SNMP_Pipeline/kafkaConsumerSNMP/src/main/java/com/example/kafkaConsumerSNMP/services/RethinkDBService.java

@@ -32,9 +32,9 @@ public class RethinkDBService {
             String jsonString = objectMapper.writeValueAsString(rethinkTrap);
             JsonNode jsonNode = objectMapper.readTree(jsonString);
             Map<String,EnrichedTrap> document = objectMapper.convertValue(jsonNode, Map.class);
-            System.out.println(jsonString);
-            System.out.println(jsonNode);
-            System.out.println(document);
+//            System.out.println(jsonString);
+//            System.out.println(jsonNode);
+//            System.out.println(document);
             r.db(connectionFactory.getDbName()).table(connectionFactory.getDbTableName()).insert(document).run(connectionFactory.getConnection());
         } catch (Exception e) {
             System.out.println("error " + e);