package kafka-eio

  1. Overview
  2. Docs
Eio-native Kafka client for OCaml 5, built on librdkafka

Install 

dune-project
 Dependency

github.com Readme Changelog Edit opam file Versions (1)

Authors

  1. L
    lbendtly

Maintainers

  1. L
    lbendtlynielsen@gmail.com

Sources

v0.1.1.tar.gz
md5=88c2b9c6d263a1eb114cc5d1b570288a
sha512=44f5395fcc58b8d4303bc0b36113a17193da3a1925125cb0742fabeb96f32eb6b243c98d2fde283b6c95872ff36ddbf96de6e98b7612799e2aefeb2f85a1db95

doc/src/kafka-eio.core/kafka_raw.ml.html

Source file kafka_raw.ml

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
type kafka_handle
type kafka_conf
type kafka_topic
type kafka_type = Producer | Consumer

external conf_new : unit -> kafka_conf
  = "ocaml_rd_kafka_conf_new"

external conf_set : kafka_conf -> string -> string -> (unit, string) result
  = "ocaml_rd_kafka_conf_set"

external kafka_new : kafka_type -> kafka_conf -> int -> (kafka_handle, string) result
  = "ocaml_rd_kafka_new"

external topic_new : kafka_handle -> string -> (kafka_topic, string) result
  = "ocaml_rd_kafka_topic_new"

external topic_destroy : kafka_topic -> unit = "ocaml_rd_kafka_topic_destroy"

external produce
  :  kafka_topic -> int32 -> bytes option -> bytes option -> int64
  -> (unit, int) result
  = "ocaml_rd_kafka_produce"

external enable_queue_events : kafka_handle -> int -> unit
  = "ocaml_kafka_enable_queue_events"

external disable_queue_events : kafka_handle -> unit
  = "ocaml_kafka_disable_queue_events"

external poll : kafka_handle -> int -> int
  = "ocaml_rd_kafka_poll"

external flush : kafka_handle -> int -> (unit, int) result
  = "ocaml_rd_kafka_flush"

(* destroy nulls the OCaml pointer before calling rd_kafka_destroy, so the GC
   finalizer becomes a no-op. Releases the domain lock during the blocking call. *)
external destroy : kafka_handle -> unit = "ocaml_rd_kafka_destroy"
let () = ignore destroy   (* suppress warning 32 — used by kafka_producer *)


external err2str : int -> string
  = "ocaml_rd_kafka_err2str"

external subscribe : kafka_handle -> string list -> (unit, string) result
  = "ocaml_rd_kafka_subscribe"

type poll_result =
  | Timeout
  | Msg of (string * int32 * int64 * bytes option * bytes option * int64 option * (string * string option) list)
  | Poll_error of int

external consumer_poll : kafka_handle -> int -> poll_result
  = "ocaml_rd_kafka_consumer_poll"

(* Event-driven consumer polling: wait on a consumer-queue wake fd, then drain
   rd_kafka_consume_queue with timeout 0. *)
external consumer_queue_events_enable : kafka_handle -> int -> unit
  = "ocaml_kafka_consumer_queue_events_enable"

external consumer_queue_events_disable : kafka_handle -> unit
  = "ocaml_kafka_consumer_queue_events_disable"

external consumer_queue_poll : kafka_handle -> int -> poll_result
  = "ocaml_rd_kafka_consumer_queue_poll"

external produce_v
  :  kafka_handle -> string -> int32 -> bytes option -> bytes option -> int64
  -> (string * string option) list
  -> (unit, int) result
  = "ocaml_rd_kafka_produce_v_bytecode" "ocaml_rd_kafka_produce_v"

external consumer_close : kafka_handle -> unit
  = "ocaml_rd_kafka_consumer_close"

external assignment_count : kafka_handle -> int
  = "ocaml_rd_kafka_assignment_count"

external assignment : kafka_handle -> (string * int32) list
  = "ocaml_rd_kafka_assignment"

external create_topic_raw : kafka_handle -> string -> int -> int -> int
  = "ocaml_rd_kafka_create_topic"
let create_topic h ~topic_name ~partitions ~replication_factor =
  create_topic_raw h topic_name partitions replication_factor

external commit_message_raw
  :  kafka_handle -> string -> int32 -> int64 -> bool
  -> (unit, int) result
  = "ocaml_rd_kafka_commit_message"
let commit_message h ~topic ~partition ~offset ~async =
  commit_message_raw h topic partition offset async

external commit_all : kafka_handle -> bool -> (unit, int) result
  = "ocaml_rd_kafka_commit_all"

external commit_offsets
  :  kafka_handle -> (string * int32 * int64) list -> bool
  -> (unit, int) result
  = "ocaml_rd_kafka_commit_offsets"

external pipe_create : unit -> int * int
  = "ocaml_kafka_pipe_create"

external delivery_sizeof : unit -> int
  = "ocaml_kafka_delivery_sizeof"
let () = ignore delivery_sizeof   (* suppress warning 32 — used by kafka_producer *)

external read_delivery : int -> int64 * int
  = "ocaml_kafka_read_delivery"

type txn_error = {
  code           : int;
  is_fatal       : bool;
  is_retriable   : bool;
  requires_abort : bool;
}

external init_transactions : kafka_handle -> int -> (unit, txn_error) result
  = "ocaml_rd_kafka_init_transactions"

external begin_transaction : kafka_handle -> (unit, txn_error) result
  = "ocaml_rd_kafka_begin_transaction"

external commit_transaction : kafka_handle -> int -> (unit, txn_error) result
  = "ocaml_rd_kafka_commit_transaction"

external abort_transaction : kafka_handle -> int -> (unit, txn_error) result
  = "ocaml_rd_kafka_abort_transaction"

external send_offsets_to_transaction
  :  kafka_handle -> kafka_handle -> (string * int32 * int64) list -> int
  -> (unit, txn_error) result
  = "ocaml_rd_kafka_send_offsets_to_transaction"

(** Pause / resume delivery for a single partition. Local operations — no
    broker round-trip. Used by consume_partitioned to stop new messages arriving
    for a partition while its retry fiber sleeps. *)
external pause_partition  : kafka_handle -> string -> int32 -> unit = "ocaml_rd_kafka_pause_partition"
external resume_partition : kafka_handle -> string -> int32 -> unit = "ocaml_rd_kafka_resume_partition"