-
Notifications
You must be signed in to change notification settings - Fork 45
/
nsq.h
127 lines (109 loc) · 4.74 KB
/
nsq.h
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
#ifndef __nsq_h
#define __nsq_h
#include <stdint.h>
#include <ev.h>
#include <evbuffsock.h>
#ifdef DEBUG
#define _DEBUG(...) fprintf(stdout, __VA_ARGS__)
#else
#define _DEBUG(...) do {;} while (0)
#endif
struct Buffer; // from evbuffsock.h
typedef enum {NSQ_FRAME_TYPE_RESPONSE, NSQ_FRAME_TYPE_ERROR, NSQ_FRAME_TYPE_MESSAGE} frame_type;
typedef enum {NSQ_PARAM_TYPE_INT, NSQ_PARAM_TYPE_CHAR} nsq_cmd_param_type;
struct NSQCmdParams {
void *v;
int t;
};
struct NSQMessage {
int64_t timestamp;
uint16_t attempts;
char id[16 + 1];
size_t body_length;
char *body;
};
struct NSQLookupdEndpoint {
char *address;
int port;
struct NSQLookupdEndpoint *next;
};
struct NSQDConnection {
char *address;
int port;
struct BufferedSocket *bs;
struct Buffer *command_buf;
uint32_t current_msg_size;
uint32_t current_frame_type;
char *current_data;
struct ev_loop *loop;
ev_timer *reconnect_timer;
void (*connect_callback)(struct NSQDConnection *conn, void *arg);
void (*close_callback)(struct NSQDConnection *conn, void *arg);
void (*msg_callback)(struct NSQDConnection *conn, struct NSQMessage *msg, void *arg);
void *arg;
struct NSQDConnection *next;
};
struct NSQReaderCfg {
ev_tstamp lookupd_interval;
size_t command_buf_len;
size_t command_buf_capacity;
size_t read_buf_len;
size_t read_buf_capacity;
size_t write_buf_len;
size_t write_buf_capacity;
};
struct NSQReader {
char *topic;
char *channel;
void *ctx; //context for call back
int max_in_flight;
struct NSQDConnection *conns;
struct NSQDConnInfo *infos;
struct NSQLookupdEndpoint *lookupd;
struct ev_timer lookupd_poll_timer;
struct ev_loop *loop;
struct NSQReaderCfg *cfg;
void *httpc;
void (*connect_callback)(struct NSQReader *rdr, struct NSQDConnection *conn);
void (*close_callback)(struct NSQReader *rdr, struct NSQDConnection *conn);
void (*msg_callback)(struct NSQReader *rdr, struct NSQDConnection *conn, struct NSQMessage *msg, void *ctx);
};
struct NSQReader *new_nsq_reader(struct ev_loop *loop, const char *topic, const char *channel, void *ctx,
struct NSQReaderCfg *cfg,
void (*connect_callback)(struct NSQReader *rdr, struct NSQDConnection *conn),
void (*close_callback)(struct NSQReader *rdr, struct NSQDConnection *conn),
void (*msg_callback)(struct NSQReader *rdr, struct NSQDConnection *conn, struct NSQMessage *msg, void *ctx));
void free_nsq_reader(struct NSQReader *rdr);
int nsq_reader_connect_to_nsqd(struct NSQReader *rdr, const char *address, int port);
int nsq_reader_connect_to_nsqlookupd(struct NSQReader *rdr);
int nsq_reader_add_nsqlookupd_endpoint(struct NSQReader *rdr, const char *address, int port);
void nsq_reader_set_loop(struct NSQReader *rdr, struct ev_loop *loop);
void nsq_run(struct ev_loop *loop);
struct NSQDConnection *new_nsqd_connection(struct ev_loop *loop, const char *address, int port,
void (*connect_callback)(struct NSQDConnection *conn, void *arg),
void (*close_callback)(struct NSQDConnection *conn, void *arg),
void (*msg_callback)(struct NSQDConnection *conn, struct NSQMessage *msg, void *arg),
void *arg);
void free_nsqd_connection(struct NSQDConnection *conn);
int nsqd_connection_connect(struct NSQDConnection *conn);
void nsqd_connection_disconnect(struct NSQDConnection *conn);
void nsqd_connection_init_timer(struct NSQDConnection *conn,
void (*reconnect_callback)(EV_P_ ev_timer *w, int revents));
void nsqd_connection_stop_timer(struct NSQDConnection *conn);
void nsq_subscribe(struct Buffer *buf, const char *topic, const char *channel);
void nsq_ready(struct Buffer *buf, int count);
void nsq_finish(struct Buffer *buf, const char *id);
void nsq_requeue(struct Buffer *buf, const char *id, int timeout_ms);
void nsq_nop(struct Buffer *buf);
void nsq_publish(struct Buffer *buf, const char *topic, const char *body);
void nsq_multi_publish(struct Buffer *buf, const char *topic, const char **body, const size_t body_size);
void nsq_defer_publish(struct Buffer *buf, const char *topic, const char *body, int defer_time_sec);
void nsq_touch(struct Buffer *buf, const char *id);
void nsq_cleanly_close_connection(struct Buffer *buf);
void nsq_auth(struct Buffer *buf, const char *secret);
void nsq_identify(struct Buffer *buf, const char *json_body);
struct NSQMessage *nsq_decode_message(const char *data, size_t data_length);
void free_nsq_message(struct NSQMessage *msg);
struct NSQLookupdEndpoint *new_nsqlookupd_endpoint(const char *address, int port);
void free_nsqlookupd_endpoint(struct NSQLookupdEndpoint *nsqlookupd_endpoint);
#endif