LDB ASYNC: core modules
[kai/samba.git] / source4 / lib / ldb / modules / paged_results.c
1 /* 
2    ldb database library
3
4    Copyright (C) Simo Sorce  2005-2008
5
6      ** NOTE! The following LGPL license applies to the ldb
7      ** library. This does NOT imply that all of Samba is released
8      ** under the LGPL
9    
10    This library is free software; you can redistribute it and/or
11    modify it under the terms of the GNU Lesser General Public
12    License as published by the Free Software Foundation; either
13    version 3 of the License, or (at your option) any later version.
14
15    This library is distributed in the hope that it will be useful,
16    but WITHOUT ANY WARRANTY; without even the implied warranty of
17    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
18    Lesser General Public License for more details.
19
20    You should have received a copy of the GNU Lesser General Public
21    License along with this library; if not, see <http://www.gnu.org/licenses/>.
22 */
23
24 /*
25  *  Name: paged_result
26  *
27  *  Component: ldb paged results control module
28  *
29  *  Description: this module caches a complete search and sends back
30  *               results in chunks as asked by the client
31  *
32  *  Author: Simo Sorce
33  */
34
35 #include "ldb_includes.h"
36
37 struct message_store {
38         /* keep the whole ldb_reply as an optimization
39          * instead of freeing and talloc-ing the container
40          * on each result */
41         struct ldb_reply *r;
42         struct message_store *next;
43 };
44
45 struct private_data;
46
47 struct results_store {
48
49         struct private_data *priv;
50
51         char *cookie;
52         time_t timestamp;
53
54         struct results_store *next;
55
56         struct message_store *first;
57         struct message_store *last;
58         int num_entries;
59
60         struct message_store *first_ref;
61         struct message_store *last_ref;
62
63         struct ldb_control **controls;
64 };
65
66 struct private_data {
67
68         int next_free_id;
69         struct results_store *store;
70         
71 };
72
73 int store_destructor(struct results_store *del)
74 {
75         struct private_data *priv = del->priv;
76         struct results_store *loop;
77
78         if (priv->store == del) {
79                 priv->store = del->next;
80                 return 0;
81         }
82
83         for (loop = priv->store; loop; loop = loop->next) {
84                 if (loop->next == del) {
85                         loop->next = del->next;
86                         return 0;
87                 }
88         }
89
90         /* is not in list ? */
91         return -1;
92 }
93
94 static struct results_store *new_store(struct private_data *priv)
95 {
96         struct results_store *newr;
97         int new_id = priv->next_free_id++;
98
99         /* TODO: we should have a limit on the number of
100          * outstanding paged searches
101          */
102
103         newr = talloc(priv, struct results_store);
104         if (!newr) return NULL;
105
106         newr->priv = priv;
107
108         newr->cookie = talloc_asprintf(newr, "%d", new_id);
109         if (!newr->cookie) {
110                 talloc_free(newr);
111                 return NULL;
112         }
113
114         newr->timestamp = time(NULL);
115
116         newr->first = NULL;
117         newr->num_entries = 0;
118         newr->first_ref = NULL;
119         newr->controls = NULL;
120
121         newr->next = priv->store;
122         priv->store = newr;
123
124         talloc_set_destructor(newr, store_destructor);
125
126         return newr;
127 }
128
129 struct paged_context {
130         struct ldb_module *module;
131         struct ldb_request *req;
132
133         struct results_store *store;
134         int size;
135         struct ldb_control **controls;
136 };
137
138 static int paged_results(struct paged_context *ac)
139 {
140         struct ldb_paged_control *paged;
141         struct message_store *msg;
142         int i, num_ctrls, ret;
143
144         if (ac->store == NULL) {
145                 return LDB_ERR_OPERATIONS_ERROR;
146         }
147
148         while (ac->store->num_entries > 0 && ac->size > 0) {
149                 msg = ac->store->first;
150                 ret = ldb_module_send_entry(ac->req, msg->r->message);
151                 if (ret != LDB_SUCCESS) {
152                         return ret;
153                 }
154
155                 ac->store->first = msg->next;
156                 talloc_free(msg);
157                 ac->store->num_entries--;
158                 ac->size--;
159         }
160
161         while (ac->store->first_ref != NULL) {
162                 msg = ac->store->first_ref;
163                 ret = ldb_module_send_referral(ac->req, msg->r->referral);
164                 if (ret != LDB_SUCCESS) {
165                         return ret;
166                 }
167
168                 ac->store->first_ref = msg->next;
169                 talloc_free(msg);
170         }
171
172         /* return result done */
173         num_ctrls = 1;
174         i = 0;
175
176         if (ac->store->controls != NULL) {
177                 while (ac->store->controls[i]) i++; /* counting */
178
179                 num_ctrls += i;
180         }
181
182         ac->controls = talloc_array(ac, struct ldb_control *, num_ctrls +1);
183         if (ac->controls == NULL) {
184                 return LDB_ERR_OPERATIONS_ERROR;
185         }
186         ac->controls[num_ctrls] = NULL;
187
188         for (i = 0; i < (num_ctrls -1); i++) {
189                 ac->controls[i] = talloc_reference(ac->controls, ac->store->controls[i]);
190         }
191
192         ac->controls[i] = talloc(ac->controls, struct ldb_control);
193         if (ac->controls[i] == NULL) {
194                 return LDB_ERR_OPERATIONS_ERROR;
195         }
196
197         ac->controls[i]->oid = talloc_strdup(ac->controls[i],
198                                                 LDB_CONTROL_PAGED_RESULTS_OID);
199         if (ac->controls[i]->oid == NULL) {
200                 return LDB_ERR_OPERATIONS_ERROR;
201         }
202
203         ac->controls[i]->critical = 0;
204
205         paged = talloc(ac->controls[i], struct ldb_paged_control);
206         if (paged == NULL) {
207                 return LDB_ERR_OPERATIONS_ERROR;
208         }
209
210         ac->controls[i]->data = paged;
211
212         if (ac->size > 0) {
213                 paged->size = 0;
214                 paged->cookie = NULL;
215                 paged->cookie_len = 0;
216         } else {
217                 paged->size = ac->store->num_entries;
218                 paged->cookie = talloc_strdup(paged, ac->store->cookie);
219                 paged->cookie_len = strlen(paged->cookie) + 1;
220         }
221
222         return LDB_SUCCESS;
223 }
224
225 static int paged_search_callback(struct ldb_request *req, struct ldb_reply *ares)
226 {
227         struct paged_context *ac ;
228         struct message_store *msg_store;
229         int ret;
230
231         ac = talloc_get_type(req->context, struct paged_context);
232
233         if (!ares) {
234                 return ldb_module_done(ac->req, NULL, NULL,
235                                         LDB_ERR_OPERATIONS_ERROR);
236         }
237         if (ares->error != LDB_SUCCESS) {
238                 return ldb_module_done(ac->req, ares->controls,
239                                         ares->response, ares->error);
240         }
241
242         switch (ares->type) {
243         case LDB_REPLY_ENTRY:
244                 msg_store = talloc(ac->store, struct message_store);
245                 if (msg_store == NULL) {
246                         return ldb_module_done(ac->req, NULL, NULL,
247                                                 LDB_ERR_OPERATIONS_ERROR);
248                 }
249                 msg_store->next = NULL;
250                 msg_store->r = talloc_steal(msg_store, ares);
251
252                 if (ac->store->first == NULL) {
253                         ac->store->first = msg_store;
254                 } else {
255                         ac->store->last->next = msg_store;
256                 }
257                 ac->store->last = msg_store;
258
259                 ac->store->num_entries++;
260
261                 break;
262
263         case LDB_REPLY_REFERRAL:
264                 msg_store = talloc(ac->store, struct message_store);
265                 if (msg_store == NULL) {
266                         return ldb_module_done(ac->req, NULL, NULL,
267                                                 LDB_ERR_OPERATIONS_ERROR);
268                 }
269                 msg_store->next = NULL;
270                 msg_store->r = talloc_steal(msg_store, ares);
271
272                 if (ac->store->first_ref == NULL) {
273                         ac->store->first_ref = msg_store;
274                 } else {
275                         ac->store->last_ref->next = msg_store;
276                 }
277                 ac->store->last_ref = msg_store;
278
279                 break;
280
281         case LDB_REPLY_DONE:
282                 ac->store->controls = talloc_move(ac->store, &ares->controls);
283                 ret = paged_results(ac);
284                 return ldb_module_done(ac->req, ac->controls,
285                                         ares->response, ret);
286         }
287
288         return LDB_SUCCESS;
289 }
290
291 static int paged_search(struct ldb_module *module, struct ldb_request *req)
292 {
293         struct ldb_control *control;
294         struct private_data *private_data;
295         struct ldb_paged_control *paged_ctrl;
296         struct ldb_control **saved_controls;
297         struct ldb_request *search_req;
298         struct paged_context *ac;
299         int ret;
300
301         /* check if there's a paged request control */
302         control = ldb_request_get_control(req, LDB_CONTROL_PAGED_RESULTS_OID);
303         if (control == NULL) {
304                 /* not found go on */
305                 return ldb_next_request(module, req);
306         }
307
308         paged_ctrl = talloc_get_type(control->data, struct ldb_paged_control);
309         if (!paged_ctrl) {
310                 return LDB_ERR_PROTOCOL_ERROR;
311         }
312
313         private_data = talloc_get_type(module->private_data, struct private_data);
314
315         ac = talloc_zero(req, struct paged_context);
316         if (ac == NULL) {
317                 ldb_set_errstring(module->ldb, "Out of Memory");
318                 return LDB_ERR_OPERATIONS_ERROR;
319         }
320
321         ac->module = module;
322         ac->req = req;
323         ac->size = paged_ctrl->size;
324
325         /* check if it is a continuation search the store */
326         if (paged_ctrl->cookie_len == 0) {
327                 if (paged_ctrl->size == 0) {
328                         return LDB_ERR_OPERATIONS_ERROR;
329                 }
330
331                 ac->store = new_store(private_data);
332                 if (ac->store == NULL) {
333                         return LDB_ERR_OPERATIONS_ERROR;
334                 }
335
336                 ret = ldb_build_search_req_ex(&search_req, module->ldb, ac,
337                                                 req->op.search.base,
338                                                 req->op.search.scope,
339                                                 req->op.search.tree,
340                                                 req->op.search.attrs,
341                                                 req->controls,
342                                                 ac,
343                                                 paged_search_callback,
344                                                 req);
345
346                 /* save it locally and remove it from the list */
347                 /* we do not need to replace them later as we
348                  * are keeping the original req intact */
349                 if (!save_controls(control, search_req, &saved_controls)) {
350                         return LDB_ERR_OPERATIONS_ERROR;
351                 }
352
353                 return ldb_next_request(module, search_req);
354
355         } else {
356                 struct results_store *current = NULL;
357
358                 /* TODO: age out old outstanding requests */
359                 for (current = private_data->store; current; current = current->next) {
360                         if (strcmp(current->cookie, paged_ctrl->cookie) == 0) {
361                                 current->timestamp = time(NULL);
362                                 break;
363                         }
364                 }
365                 if (current == NULL) {
366                         return LDB_ERR_UNWILLING_TO_PERFORM;
367                 }
368
369                 ac->store = current;
370
371                 /* check if it is an abandon */
372                 if (ac->size == 0) {
373                         return ldb_module_done(req, NULL, NULL,
374                                                                 LDB_SUCCESS);
375                 }
376
377                 ret = paged_results(ac);
378                 if (ret != LDB_SUCCESS) {
379                         return ldb_module_done(req, NULL, NULL, ret);
380                 }
381                 return ldb_module_done(req, ac->controls, NULL,
382                                                                 LDB_SUCCESS);
383         }
384 }
385
386 static int paged_request_init(struct ldb_module *module)
387 {
388         struct private_data *data;
389         int ret;
390
391         data = talloc(module, struct private_data);
392         if (data == NULL) {
393                 return LDB_ERR_OTHER;
394         }
395
396         data->next_free_id = 1;
397         data->store = NULL;
398         module->private_data = data;
399
400         ret = ldb_mod_register_control(module, LDB_CONTROL_PAGED_RESULTS_OID);
401         if (ret != LDB_SUCCESS) {
402                 ldb_debug(module->ldb, LDB_DEBUG_WARNING,
403                         "paged_request:"
404                         "Unable to register control with rootdse!\n");
405         }
406
407         return ldb_next_init(module);
408 }
409
410 const struct ldb_module_ops ldb_paged_results_module_ops = {
411         .name           = "paged_results",
412         .search         = paged_search,
413         .init_context   = paged_request_init
414 };