hook kea con amqp

1.4.0/kea-ik/Dockerfile

@@ -18,6 +18,9 @@ RUN wget -O libzdb-3.1.tar.gz http://www.tildeslash.com/libzdb/dist/libzdb-3.1.t
 RUN tar xvzf libzdb-3.1.tar.gz
 RUN cd libzdb-3.1 && ./configure --without-postgresql --without-sqlite && make && make install
 
+RUN git clone https://github.com/alanxz/SimpleAmqpClient
+RUN cd SimpleAmqpClient && mkdir simpleamqpclient-build && cd simpleamqpclient-build && cmake .. && make && make install
+
 RUN ldconfig
 
 # FLOWDAT KEA HOOK

hooks/amqp/Makefile

@@ -5,9 +5,9 @@ KEA_LIB ?= /opt/kea/lib/
 
 OBJECTS = src/messages.o src/logger.o src/load.o src/callouts.o src/version.o src/amqppublisher.o
 DEPS = $(OBJECTS:.o=.d)
-CXXFLAGS = -I $(KEA_INCLUDE) -g -fPIC -Wno-deprecated -std=c++11
+CXXFLAGS = -I $(KEA_INCLUDE) -I/SimpleAmqpClient/src -g -fPIC -Wno-deprecated -std=c++11
 LDFLAGS = -L $(KEA_LIB) -shared
-LIBRARIES = -lkea-dhcpsrv -lkea-dhcp++ -lkea-hooks -lkea-log -lkea-util -lkea-exceptions -lamqpcpp -lrabbitmq 
+LIBRARIES = -lkea-dhcpsrv -lkea-dhcp++ -lkea-hooks -lkea-log -lkea-util -lkea-exceptions -lamqpcpp -lrabbitmq -lSimpleAmqpClient 
 
 kea-hook-flowdat3.so: $(OBJECTS)
 	$(CXX) -o $@ $(CXXFLAGS) $(LDFLAGS) $(OBJECTS) $(LIBRARIES)

hooks/amqp/src/amqppublisher.cc

@@ -1,32 +1,26 @@
-#include <AMQPcpp.h>
-#include "amqppublisher.h"
-
+#include <unistd.h>
 #include <string>
+#include "amqppublisher.h"
+#include <SimpleAmqpClient/SimpleAmqpClient.h>
 
 int AMQPPublisher::publish(std::string message)
 {
-    AMQP amqp("amqp:5672");
-    try {
-        AMQPExchange * ex = amqp.createExchange("kea");
-    
-        ex->Declare("kea", "fanout");
-    
-        AMQPQueue * qu2 = amqp.createQueue("kea");
-        
-        qu2->Declare();
-        qu2->Bind("kea", "");
-        
-        ex->setHeader("Expiration", "10000");
-        ex->setHeader("Delivery-mode", "2");
-        ex->setHeader("Content-type", "text/text");
-        ex->setHeader("Content-encoding", "UTF-8");
-    
-        ex->Publish(message, "");				
-    } catch (AMQPException e) {
-        std::cout << e.getMessage() << std::endl;
-    }
-    
-    amqp.closeChannel();
+    using namespace AmqpClient;
+
+    Channel::ptr_t channel = Channel::Create("amqp", 5672, "guest", "guest", "/");
+
+    BasicMessage::ptr_t basic_message = BasicMessage::Create(message);
+    basic_message->Expiration("10000");
+    basic_message->DeliveryMode(BasicMessage::dm_persistent);
+    basic_message->ContentType("text/text");
+    basic_message->ContentEncoding("UTF-8");
     
+    channel->DeclareExchange("kea", "fanout");
+
+    std::string queue = channel->DeclareQueue("kea", false, false, false, false);
+    channel->BindQueue(queue, "kea");
+
+    channel->BasicPublish("kea", queue, basic_message, false, false);
+
     return 0;
 }

hooks/amqp/src/callouts.cc

@@ -64,7 +64,7 @@ int subnet4_select(CalloutHandle& handle) {
     handle.getArgument("subnet4", subnet);
     
     amqp.publish(query->toText());
-    amqp.publish(subnet->toText());
+    amqp.publish("subnet4_select: " + subnet->toText());
     
     return 0;
 }
@@ -81,8 +81,8 @@ int lease4_select(CalloutHandle& handle) {
     handle.getArgument("lease4", lease);
     
     amqp.publish(query->toText());
-    amqp.publish(subnet->toText());
-    amqp.publish(lease->toText());
+    amqp.publish("lease4_select: " + subnet->toText());
+    amqp.publish("lease4_select: " + lease->toText());
     
     return 0;
 }
@@ -97,8 +97,8 @@ int lease4_renew(CalloutHandle& handle) {
     handle.getArgument("lease4", lease);
     
     amqp.publish(query->toText());
-    amqp.publish(subnet->toText());
-    amqp.publish(lease->toText());
+    amqp.publish("lease4_renew: " + subnet->toText());
+    amqp.publish("lease4_renew: " + lease->toText());
     
     return 0;
 }
@@ -111,7 +111,7 @@ int lease4_release(CalloutHandle& handle) {
     handle.getArgument("lease4", lease);
     
     amqp.publish(query->toText());
-    amqp.publish(lease->toText());
+    amqp.publish("lease4_release: " + lease->toText());
     
     return 0;
 }
@@ -124,7 +124,7 @@ int lease4_decline(CalloutHandle& handle) {
     handle.getArgument("lease4", lease);
     
     amqp.publish(query->toText());
-    amqp.publish(lease->toText());
+    amqp.publish("lease4_decline: " + lease->toText());
     
     return 0;
 }
@@ -136,7 +136,7 @@ int lease4_expire(CalloutHandle& handle) {
     handle.getArgument("lease4", lease);
     handle.getArgument("remove_lease", remove_lease);
     
-    amqp.publish(lease->toText());
+    amqp.publish("lease4_expire: " + lease->toText());
     
     return 0;
 }
@@ -146,7 +146,7 @@ int lease4_recover(CalloutHandle& handle) {
     Lease4Ptr lease;
     handle.getArgument("lease4", lease);
     
-    amqp.publish(lease->toText());
+    amqp.publish("lease4_recover: " + lease->toText());
     
     return 0;
 }
@@ -182,7 +182,7 @@ int subnet6_select(CalloutHandle& handle) {
     handle.getArgument("subnet6", subnet);
     
     amqp.publish(query->toText());
-    amqp.publish(subnet->toText());
+    amqp.publish("subnet6_select: " + subnet->toText());
     
     return 0;
 }
@@ -199,8 +199,8 @@ int lease6_select(CalloutHandle& handle) {
     handle.getArgument("lease6", lease);
     
     amqp.publish(query->toText());
-    amqp.publish(subnet->toText());
-    amqp.publish(lease->toText());
+    amqp.publish("lease6_select: " + subnet->toText());
+    amqp.publish("lease6_select: " + lease->toText());
     
     return 0;
 }
@@ -215,16 +215,16 @@ int lease6_renew(CalloutHandle& handle) {
     handle.getArgument("lease6", lease);
     
     amqp.publish(query->toText());
-    amqp.publish(lease->toText());
+    amqp.publish("lease6_renew: " + lease->toText());
     
     try {
         handle.getArgument("ia_na", ia_na);
-        amqp.publish(ia_na->toText());
+        amqp.publish("lease6_renew: " + ia_na->toText());
     } catch (const NoSuchArgument&) { }
     
     try {
         handle.getArgument("ia_pd", ia_pd);
-        amqp.publish(ia_pd->toText());
+        amqp.publish("lease6_renew: " + ia_pd->toText());
     } catch (const NoSuchArgument&) { }
     
     return 0;
@@ -240,16 +240,16 @@ int lease6_rebind(CalloutHandle& handle) {
     handle.getArgument("lease6", lease);
     
     amqp.publish(query->toText());
-    amqp.publish(lease->toText());
+    amqp.publish("lease6_rebind: " + lease->toText());
     
     try {
         handle.getArgument("ia_na", ia_na);
-        amqp.publish(ia_na->toText());
+        amqp.publish("lease6_rebind: " + ia_na->toText());
     } catch (const NoSuchArgument&) { }
     
     try {
         handle.getArgument("ia_pd", ia_pd);
-        amqp.publish(ia_pd->toText());
+        amqp.publish("lease6_rebind: " + ia_pd->toText());
     } catch (const NoSuchArgument&) { }
     
     return 0;
@@ -263,7 +263,7 @@ int lease6_decline(CalloutHandle& handle) {
     handle.getArgument("lease6", lease);
     
     amqp.publish(query->toText());
-    amqp.publish(lease->toText());
+    amqp.publish("lease6_decline: " + lease->toText());
     
     return 0;
 }
@@ -276,7 +276,7 @@ int lease6_release(CalloutHandle& handle) {
     handle.getArgument("lease6", lease);
     
     amqp.publish(query->toText());
-    amqp.publish(lease->toText());
+    amqp.publish("lease6_release: " + lease->toText());
     
     return 0;
 }
@@ -288,7 +288,7 @@ int lease6_expire(CalloutHandle& handle) {
     handle.getArgument("lease6", lease);
     handle.getArgument("remove_lease", remove_lease);
     
-    amqp.publish(lease->toText());
+    amqp.publish("lease6_expire: " + lease->toText());
     
     return 0;
 }
@@ -298,7 +298,7 @@ int lease6_recover(CalloutHandle& handle) {
     Lease6Ptr lease;
     handle.getArgument("lease6", lease);
     
-    amqp.publish(lease->toText());
+    amqp.publish("lease6_recover: " + lease->toText());
     
     return 0;
 }