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
|
/*
* Copyright (c) 2023 One Identity LLC.
* Copyright (c) 2020 Balabit
*
* This program is free software; you can redistribute it and/or modify it
* under the terms of the GNU General Public License version 2 as published
* by the Free Software Foundation, or (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program; if not, write to the Free Software
* Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
*
* As an additional exemption you are allowed to compile & link against the
* OpenSSL libraries as published by the OpenSSL project. See the file
* COPYING for details.
*
*/
#include "example_destination_worker.h"
#include "example_destination.h"
#include "thread-utils.h"
#include <stdio.h>
static LogThreadedResult
_dw_insert(LogThreadedDestWorker *s, LogMessage *msg)
{
ExampleDestinationWorker *self = (ExampleDestinationWorker *)s;
GString *string_to_write = g_string_new("");
g_string_printf(string_to_write, "thread_id=%lu message=%s\n",
(unsigned long) self->thread_id, log_msg_get_value(msg, LM_V_MESSAGE, NULL));
size_t retval = fwrite(string_to_write->str, 1, string_to_write->len, self->file);
gsize string_to_write_len = string_to_write->len;
g_string_free(string_to_write, TRUE);
if (retval != string_to_write_len)
{
msg_error("Error while reading file");
return LTR_NOT_CONNECTED;
}
if (fflush(self->file) != 0)
{
msg_error("Error while flushing file");
return LTR_NOT_CONNECTED;
}
return LTR_SUCCESS;
/*
* LTR_DROP,
* LTR_ERROR,
* LTR_SUCCESS,
* LTR_QUEUED,
* LTR_NOT_CONNECTED,
* LTR_RETRY,
*/
}
static gboolean
_connect(LogThreadedDestWorker *s)
{
ExampleDestinationWorker *self = (ExampleDestinationWorker *)s;
ExampleDestinationDriver *owner = (ExampleDestinationDriver *) s->owner;
self->file = fopen(owner->filename->str, "a");
if (!self->file)
{
msg_error("Could not open file", evt_tag_error("error"));
return FALSE;
}
return TRUE;
}
static void
_disconnect(LogThreadedDestWorker *s)
{
ExampleDestinationWorker *self = (ExampleDestinationWorker *)s;
fclose(self->file);
}
static gboolean
_dw_init(LogThreadedDestWorker *s)
{
ExampleDestinationWorker *self = (ExampleDestinationWorker *)s;
/*
You can create thread specific resources here. In this example, we
store the thread id.
*/
self->thread_id = get_thread_id();
return log_threaded_dest_worker_init_method(s);
}
static void
_dw_deinit(LogThreadedDestWorker *s)
{
/*
If you created resources during _thread_init,
you need to free them here
*/
log_threaded_dest_worker_deinit_method(s);
}
static void
_dw_free(LogThreadedDestWorker *s)
{
/*
If you created resources during new,
you need to free them here.
*/
log_threaded_dest_worker_free_method(s);
}
LogThreadedDestWorker *
example_destination_dw_new(LogThreadedDestDriver *o, gint worker_index)
{
ExampleDestinationWorker *self = g_new0(ExampleDestinationWorker, 1);
log_threaded_dest_worker_init_instance(&self->super, o, worker_index);
self->super.init = _dw_init;
self->super.deinit = _dw_deinit;
self->super.insert = _dw_insert;
self->super.free_fn = _dw_free;
self->super.connect = _connect;
self->super.disconnect = _disconnect;
return &self->super;
}
|
