package kafka-eio

  1. Overview
  2. Docs

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"