Port to jbuilder.

bin/jbuild

@@ -0,0 +1,12 @@
+(jbuild_version 1)
+
+(executables
+ ((names (sendto_kafka_topic tail_kafka_topic))
+  (libraries (okafka.lwt cmdliner))))
+
+(install
+ ((section bin)
+  (files (
+   (sendto_kafka_topic.exe as sendto_kafka_topic)
+   (tail_kafka_topic.exe as tail_kafka_topic)
+  ))))

lib/jbuild

@@ -0,0 +1,7 @@
+(jbuild_version 1)
+
+(library
+ ((name kafka)
+  (public_name okafka)
+  (c_names (ocaml_kafka))
+  (c_library_flags (-lrdkafka -lpthread -lz))))

lib_helpers/jbuild

@@ -0,0 +1,6 @@
+(jbuild_version 1)
+
+(library
+ ((name kafka_helpers)
+  (public_name okafka.helpers)
+  (libraries (okafka))))

lib_lwt/jbuild

@@ -0,0 +1,8 @@
+(jbuild_version 1)
+
+(library
+ ((name kafka_lwt)
+  (public_name okafka.lwt)
+  (libraries (okafka lwt.unix))
+  (c_names (ocaml_lwt_kafka))
+  (c_library_flags (-lrdkafka -lpthread -lz))))

okafka.opam

@@ -0,0 +1,16 @@
+opam-version: "1.2"
+version: "0.3.0"
+name: "okafka"
+synopsis: "OCaml bindings for Kafka"
+maintainer: "Didier Wenzek <[email protected]>"
+authors: ["Didier Wenzek <[email protected]>"]
+homepage: "https://github.com/didier-wenzek/ocaml-kafka"
+bug-reports: "https://github.com/didier-wenzek/ocaml-kafka/issues"
+dev-repo: "https://github.com/didier-wenzek/ocaml-kafka"
+license: "GPL"
+build: [
+  ["jbuilder" "build" "--only" "okafka" "--root" "." "-j" jobs "@install"]
+]
+install: [make "install"]
+remove: [make "uninstall"]
+depends: ["jbuilder"]

test/jbuild

@@ -0,0 +1,10 @@
+(jbuild_version 1)
+
+(executables
+ ((names (tests issue3))
+  (libraries (okafka.lwt okafka.helpers))))
+
+(alias
+ ((name    runtest)
+  (deps    (tests.exe))
+  (action  (run ${<}))))

tests.ml → test/tests.ml

@@ -61,7 +61,7 @@ let main =
    let rec consume t p = match Kafka.consume ~timeout_ms t p with
       | Kafka.Message(_,_,_,msg,_) -> msg
       | Kafka.PartitionEnd(_,_,_) -> (
-          Printf.fprintf stderr "No message for now\n%!";
+          (* Printf.fprintf stderr "No message for now\n%!"; *)
           consume t p
       )
       | exception Kafka.Error(Kafka.TIMED_OUT,_) -> (
@@ -112,7 +112,7 @@ let main =
           else (n,m+1)
       )
       | Kafka.PartitionEnd(_,_,_) -> (
-          Printf.fprintf stderr "No queued message for now\n%!";
+          (* Printf.fprintf stderr "No queued message for now\n%!"; *)
           consume_queue (n,m)
       )
       | exception Kafka.Error(Kafka.TIMED_OUT,_) -> (
@@ -147,7 +147,7 @@ let main =
    let rec consume_k t = match Kafka.consume_queue t with
       | Kafka.Message(_,_,_,msg,key) -> key,msg
       | Kafka.PartitionEnd(_,_,_) -> (
-          Printf.fprintf stderr "No keyed message for now\n%!";
+          (* Printf.fprintf stderr "No keyed message for now\n%!"; *)
           consume_k t
       )
       | exception Kafka.Error(Kafka.TIMED_OUT,_) -> (
@@ -183,15 +183,20 @@ let main =
    Kafka.destroy_handler consumer;
 
    (* KafkaConsumer / KafkaProducer API *)
+   let open Kafka_helpers in
    let stop_at_end = true in
    let iterable_of_list xs f = List.iter f xs in
-   let sink = KafkaProducer.partition_sink "test" Kafka.partition_unassigned in
-   let src = KafkaConsumer.fold_topic ~stop_at_end "test" [] in
+   let sink = Kafka_producer.partition_sink "test" Kafka.partition_unassigned in
+   let src = Kafka_consumer.fold_topic ~stop_at_end "test" [] in
    let offsets = [0,Kafka.offset_tail 3; 1,Kafka.offset_tail 3] in
 
-   ["message 123"; "message 124"; "message 125"] |> iterable_of_list |> KafkaProducer.stream_to sink;
+   ["message 123"; "message 124"; "message 125"] |> iterable_of_list |> Kafka_producer.stream_to sink;
    let messages = src (fun acc -> function Kafka.Message(_,_,_,msg,_) -> msg::acc | _ -> acc) offsets [] in
    assert (List.length messages = 6);
    assert (List.exists (fun msg -> msg = "message 123") messages);
    assert (List.exists (fun msg -> msg = "message 124") messages);
-   assert (List.exists (fun msg -> msg = "message 125") messages)
+   assert (List.exists (fun msg -> msg = "message 125") messages);
+
+
+   "Tests successful\n%!"
+