Nyx Node
Loading...
Searching...
No Matches
indi_stream.c
1/* NyxNode
2 * Author: Jérôme ODIER <jerome.odier@lpsc.in2p3.fr>
3 * SPDX-License-Identifier: GPL-2.0-only (Mongoose backend) or GPL-3.0+
4 */
5
6/*--------------------------------------------------------------------------------------------------------------------*/
7
8#include <stdio.h>
9#include <string.h>
10
11#include "../nyx_node_internal.h"
12
13/*--------------------------------------------------------------------------------------------------------------------*/
14/* PROP */
15/*--------------------------------------------------------------------------------------------------------------------*/
16
17nyx_dict_t *nyx_stream_prop_new(STR_t name, STR_t label)
18{
19 if(label == NULL || label[0] == '\0')
20 {
21 label = name;
22 }
23
24 /*----------------------------------------------------------------------------------------------------------------*/
25
26 char hash[8 + 1];
27
28 snprintf(hash, sizeof(hash), "%08X", nyx_hash(
29 strlen(name),
30 buffof(name),
31 NYX_STREAM_MAGIC
32 ));
33
34 /*----------------------------------------------------------------------------------------------------------------*/
35
36 nyx_dict_t *result = nyx_dict_new();
37
38 /*----------------------------------------------------------------------------------------------------------------*/
39
40 nyx_dict_set_string_unref(result, "<>", "defStream", false);
41
42 /*----------------------------------------------------------------------------------------------------------------*/
43
44 nyx_dict_set_string_unref(result, "@name", nyx_string_dup(name), true);
45 nyx_dict_set_string_unref(result, "@hash", nyx_string_dup(hash), true);
46 nyx_dict_set_string_unref(result, "@label", nyx_string_dup(label), true);
47
48 /*----------------------------------------------------------------------------------------------------------------*/
49
50 return result;
51}
52
53/*--------------------------------------------------------------------------------------------------------------------*/
54/* VECTOR */
55/*--------------------------------------------------------------------------------------------------------------------*/
56
57nyx_dict_t *nyx_stream_vector_new(
58 STR_t device,
59 STR_t name,
60 nyx_state_t state,
61 nyx_dict_t *props[],
62 const nyx_opts_t *opts
63) {
64 nyx_dict_t *result = nyx_dict_new();
65
66 nyx_list_t *children = nyx_list_new();
67
68 /*----------------------------------------------------------------------------------------------------------------*/
69
70 nyx_dict_set_string_unref(result, "<>", "defStreamVector", false);
71
72 /*----------------------------------------------------------------------------------------------------------------*/
73
74 nyx_dict_set_string_unref(result, "@client", "unknown", false);
75 nyx_dict_set_string_unref(result, "@device", nyx_string_dup(device), true);
76 nyx_dict_set_string_unref(result, "@name", nyx_string_dup(name), true);
77
78 nyx_dict_set_string_unref(result, "@state", nyx_state_to_str(state), false);
79
80 /*----------------------------------------------------------------------------------------------------------------*/
81
82 internal_set_opts(result, opts);
83
84 /*----------------------------------------------------------------------------------------------------------------*/
85
86 nyx_dict_set(result, "children", children);
87
88 if(props) for(; *props != NULL; props++)
89 {
90 nyx_list_push(children, *props);
91
92 nyx_object_unref(*props);
93 }
94
95 nyx_object_unref(children);
96
97 /*----------------------------------------------------------------------------------------------------------------*/
98
99 return result;
100}
101
102/*--------------------------------------------------------------------------------------------------------------------*/
103/* SET VECTOR */
104/*--------------------------------------------------------------------------------------------------------------------*/
105
106nyx_dict_t *nyx_stream_set_vector_new(const nyx_dict_t *vector)
107{
108 return internal_prop_to_set_vector(vector, "setStreamVector", "oneStream");
109}
110
111/*--------------------------------------------------------------------------------------------------------------------*/
112/* PUBLISHER */
113/*--------------------------------------------------------------------------------------------------------------------*/
114
115bool nyx_stream_pub(const nyx_dict_t *vector, size_t n_fields, const size_t field_sizes[], const buff_t field_buffs[])
116{
117 /*----------------------------------------------------------------------------------------------------------------*/
118 /* CHECK IF STREAM IS ENABLED OR EMPTY */
119 /*----------------------------------------------------------------------------------------------------------------*/
120
121 if((vector->base.flags & NYX_FLAGS_STREAM_MASK) == 0 || n_fields == 0)
122 {
123 return true;
124 }
125
126 /*----------------------------------------------------------------------------------------------------------------*/
127 /* RETRIEVE INFORMATION */
128 /*----------------------------------------------------------------------------------------------------------------*/
129
130 const nyx_node_t *node = vector->base.node;
131
132 STR_t device = nyx_dict_get_string(vector, "@device");
133 STR_t stream = nyx_dict_get_string(vector, "@name" );
134
135 if(node == NULL || device == NULL || stream == NULL)
136 {
137 NYX_LOG_ERROR("Stream vector not properly initialized");
138
139 return false;
140 }
141
142 /*----------------------------------------------------------------------------------------------------------------*/
143 /* PREPROCESS DATA */
144 /*----------------------------------------------------------------------------------------------------------------*/
145
146 size_t idx = 0;
147
148 nyx_object_t *dict;
149
150 uint32_t prepd_hashes[n_fields];
151
152 size_t prepd_sizes[n_fields];
153 BUFF_t prepd_buffs[n_fields];
154
155 /*----------------------------------------------------------------------------------------------------------------*/
156
157 nyx_object_t *list = nyx_dict_get(vector, "children");
158
159 if(list != NULL && list->type == NYX_TYPE_LIST)
160 {
161 /*------------------------------------------------------------------------------------------------------------*/
162
163 for(nyx_list_iter_t iter = NYX_LIST_ITER(list); nyx_list_iterate(&iter, &idx, &dict);)
164 {
165 if(dict->type == NYX_TYPE_DICT && idx < n_fields)
166 {
167 nyx_object_t *string = nyx_dict_get((nyx_dict_t *) dict, "@name");
168
169 if(string != NULL && string->type == NYX_TYPE_STRING)
170 {
171 /*------------------------------------------------------------------------------------------------*/
172
173 STR_t field_name_buf = nyx_string_get((nyx_string_t *) string);
174
175 size_t field_name_len = strlen(field_name_buf);
176
177 size_t field_size = field_sizes[idx];
178 BUFF_t field_buff = field_buffs[idx];
179
180 /*------------------------------------------------------------------------------------------------*/
181 /* COMPUTE HASH */
182 /*------------------------------------------------------------------------------------------------*/
183
184 prepd_hashes[idx] = nyx_hash(field_name_len, field_name_buf, NYX_STREAM_MAGIC);
185
186 /*------------------------------------------------------------------------------------------------*/
187 /* BASE64 PAYLOAD */
188 /*------------------------------------------------------------------------------------------------*/
189
190 /**/ if(field_name_len > 2 && field_name_buf[field_name_len - 2] == '.' && field_name_buf[field_name_len - 1] == 'b')
191 {
192 prepd_buffs[idx] = nyx_base64_encode(&prepd_sizes[idx], field_size, field_buff);
193 }
194
195 /*------------------------------------------------------------------------------------------------*/
196 /* ZLIB PAYLOAD */
197 /*------------------------------------------------------------------------------------------------*/
198
199 else if(field_name_len > 2 && field_name_buf[field_name_len - 2] == '.' && field_name_buf[field_name_len - 1] == 'z')
200 {
201 prepd_buffs[idx] = nyx_zlib_deflate(&prepd_sizes[idx], field_size, field_buff);
202 }
203
204 /*------------------------------------------------------------------------------------------------*/
205 /* RAW PAYLOAD */
206 /*------------------------------------------------------------------------------------------------*/
207
208 else
209 {
210 prepd_sizes[idx] = (size_t) /* NOSONAR */ field_size;
211 prepd_buffs[idx] = (buff_t) /* NOSONAR */ field_buff;
212 }
213
214 /*------------------------------------------------------------------------------------------------*/
215 }
216 else
217 {
218 NYX_LOG_ERROR("Invalid stream property");
219
220 return false;
221 }
222 }
223 else
224 {
225 NYX_LOG_ERROR("Invalid stream property");
226
227 return false;
228 }
229 }
230
231 /*------------------------------------------------------------------------------------------------------------*/
232 }
233 else
234 {
235 NYX_LOG_ERROR("Invalid stream vector");
236
237 return false;
238 }
239
240 /*----------------------------------------------------------------------------------------------------------------*/
241
242 if(n_fields != idx + 1)
243 {
244 NYX_LOG_ERROR("Missing fields, %d expected, %d provided", (int) n_fields, (int) idx + 1);
245
246 return false;
247 }
248
249 /*----------------------------------------------------------------------------------------------------------------*/
250 /* PUBLISH STREAM */
251 /*----------------------------------------------------------------------------------------------------------------*/
252
253 nyx_nss_pub(node, device, stream, n_fields, prepd_hashes, prepd_sizes, prepd_buffs);
254
255 /*----------------------------------------------------------------------------------------------------------------*/
256
257 for(size_t i = 0; i < n_fields; i++)
258 {
259 if(prepd_buffs[i] != field_buffs[i])
260 {
261 nyx_memory_free((buff_t) /* NOSONAR */ prepd_buffs[i]);
262 }
263 }
264
265 /*----------------------------------------------------------------------------------------------------------------*/
266
267 return true;
268}
269
270/*--------------------------------------------------------------------------------------------------------------------*/
nyx_dict_t
Struct describing a JSON dict object.
NYX_LIST_ITER
#define NYX_LIST_ITER(list)
Initializes a JSON list iterator.
Definition nyx_node.h:1457
nyx_list_iter_t
Struct describing a JSON list iterator.
Definition nyx_node.h:1443
nyx_list_t
Struct describing a JSON list object.
NYX_LOG_ERROR
#define NYX_LOG_ERROR(fmt,...)
Logs an error message.
Definition nyx_node.h:219
STR_t
#define STR_t
Alias for const char *.
Definition nyx_node.h:71
nyx_memory_free
__NYX_ZEROABLE__ size_t nyx_memory_free(__NYX_NULLABLE__ buff_t buff)
Similar to libc free except that it returns the amount of memory freed.
buff_t
#define buff_t
Alias for void *.
Definition nyx_node.h:67
BUFF_t
#define BUFF_t
Alias for const void *.
Definition nyx_node.h:68
nyx_string_dup
__NYX_NULLABLE__ str_t nyx_string_dup(__NYX_NULLABLE__ STR_t s)
Similar to libc strdup.
nyx_node_t
Opaque struct describing a Nyx node.
nyx_state_t
nyx_state_t
Vector state hint.
Definition nyx_node.h:1913
nyx_opts_t
Struct describing the options for INDI / Nyx vectors.
Definition nyx_node.h:2049
NYX_TYPE_DICT
@ NYX_TYPE_DICT
Dict object.
Definition nyx_node.h:438
NYX_TYPE_LIST
@ NYX_TYPE_LIST
List object.
Definition nyx_node.h:439
NYX_TYPE_STRING
@ NYX_TYPE_STRING
String object.
Definition nyx_node.h:437
nyx_object_t
Struct describing a JSON object.
nyx_string_t
Struct describing a JSON string object.
Definition nyx_node.h:914
nyx_hash
uint32_t nyx_hash(__NYX_ZEROABLE__ size_t size, __NYX_NULLABLE__ BUFF_t buff, uint32_t seed)
Hashes a buffer using the MurmurHash2 algorithm.
nyx_zlib_deflate
__NYX_NULLABLE__ buff_t nyx_zlib_deflate(__NYX_NULLABLE__ size_t *result_size, __NYX_ZEROABLE__ size_t size, __NYX_NULLABLE__ BUFF_t buff)
Compresses a buffer using the ZLib algorithm.
nyx_base64_encode
__NYX_NULLABLE__ str_t nyx_base64_encode(__NYX_NULLABLE__ size_t *result_len, __NYX_ZEROABLE__ size_t size, __NYX_NULLABLE__ BUFF_t buff)
Encodes a buffer using the Base64 algorithm.