[kdbus] Integrate kdbus core with new API.
[platform/upstream/glib.git] / gio / tests / gdbus-connection-flush.c
1 /* Test case for GNOME #662395
2  *
3  * Copyright (C) 2008-2010 Red Hat, Inc.
4  * Copyright (C) 2011 Nokia Corporation
5  *
6  * This library is free software; you can redistribute it and/or
7  * modify it under the terms of the GNU Lesser General Public
8  * License as published by the Free Software Foundation; either
9  * version 2 of the License, or (at your option) any later version.
10  *
11  * This library is distributed in the hope that it will be useful,
12  * but WITHOUT ANY WARRANTY; without even the implied warranty of
13  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
14  * Lesser General Public License for more details.
15  *
16  * You should have received a copy of the GNU Lesser General
17  * Public License along with this library; if not, see <http://www.gnu.org/licenses/>.
18  *
19  * Author: Simon McVittie <simon.mcvittie@collabora.co.uk>
20  */
21
22 #include <config.h>
23
24 #include <unistd.h>
25 #include <string.h>
26
27 #include <gio/gio.h>
28
29 #include "test-io-stream.h"
30 #include "test-pipe-unix.h"
31
32 #define MY_TYPE_OUTPUT_STREAM \
33         (my_output_stream_get_type ())
34 #define MY_OUTPUT_STREAM(o) \
35         (G_TYPE_CHECK_INSTANCE_CAST ((o), \
36                                      MY_TYPE_OUTPUT_STREAM, \
37                                      MyOutputStream))
38 #define MY_IS_OUTPUT_STREAM(o) \
39         (G_TYPE_CHECK_INSTANCE_TYPE ((o), MY_TYPE_OUTPUT_STREAM))
40
41 G_LOCK_DEFINE_STATIC (write);
42
43 typedef struct {
44     GFilterOutputStream parent;
45
46     volatile gint started;
47     volatile gint finished;
48     volatile gint flushed;
49
50     GOutputStream *real_output;
51 } MyOutputStream;
52
53 typedef struct {
54     GFilterOutputStreamClass parent;
55 } MyOutputStreamClass;
56
57 static GType my_output_stream_get_type (void) G_GNUC_CONST;
58
59 G_DEFINE_TYPE (MyOutputStream, my_output_stream, G_TYPE_FILTER_OUTPUT_STREAM)
60
61 /* Called from GDBusWorker thread */
62 static gssize
63 my_output_stream_write (GOutputStream  *os,
64                         const void     *buffer,
65                         gsize           count,
66                         GCancellable   *cancellable,
67                         GError        **error)
68 {
69   MyOutputStream *self = MY_OUTPUT_STREAM (os);
70   GFilterOutputStream *filter = G_FILTER_OUTPUT_STREAM (os);
71   GOutputStream *real = g_filter_output_stream_get_base_stream (filter);
72   gssize ret;
73
74   g_atomic_int_add (&self->started, count);
75   /* Other threads can make writing block forever by taking this lock */
76   G_LOCK (write);
77   ret = g_output_stream_write (real, buffer, count, cancellable, error);
78   G_UNLOCK (write);
79   g_atomic_int_add (&self->finished, count);
80   return ret;
81 }
82
83 /* Called from GDBusWorker thread */
84 static gboolean
85 my_output_stream_flush (GOutputStream             *os,
86                         GCancellable              *cancellable,
87                         GError                   **error)
88 {
89   MyOutputStream *self = MY_OUTPUT_STREAM (os);
90   GFilterOutputStream *filter = G_FILTER_OUTPUT_STREAM (os);
91   GOutputStream *real = g_filter_output_stream_get_base_stream (filter);
92   gint started, finished;
93   gboolean ret;
94
95   /* These should be equal because you're not allowed to flush with a
96    * write pending, and GOutputStream enforces that for its subclasses
97    */
98   started = g_atomic_int_get (&self->started);
99   finished = g_atomic_int_get (&self->finished);
100   g_assert_cmpint (started, ==, finished);
101
102   ret = g_output_stream_flush (real, cancellable, error);
103
104   /* As above, this shouldn't have changed during the flush */
105   finished = g_atomic_int_get (&self->finished);
106   g_assert_cmpint (started, ==, finished);
107
108   /* Checkpoint reached */
109   g_atomic_int_set (&self->flushed, finished);
110   return ret;
111 }
112
113 /* Called from any thread; thread-safe */
114 static gint
115 my_output_stream_get_bytes_started (GOutputStream *os)
116 {
117   MyOutputStream *self = MY_OUTPUT_STREAM (os);
118
119   return g_atomic_int_get (&self->started);
120 }
121
122 /* Called from any thread; thread-safe */
123 static gint
124 my_output_stream_get_bytes_finished (GOutputStream *os)
125 {
126   MyOutputStream *self = MY_OUTPUT_STREAM (os);
127
128   return g_atomic_int_get (&self->finished);
129 }
130
131 /* Called from any thread; thread-safe */
132 static gint
133 my_output_stream_get_bytes_flushed (GOutputStream *os)
134 {
135   MyOutputStream *self = MY_OUTPUT_STREAM (os);
136
137   return g_atomic_int_get (&self->flushed);
138 }
139
140 static void
141 my_output_stream_init (MyOutputStream *self)
142 {
143 }
144
145 static void
146 my_output_stream_class_init (MyOutputStreamClass *cls)
147 {
148   GOutputStreamClass *ostream_class = (GOutputStreamClass *) cls;
149
150   ostream_class->write_fn = my_output_stream_write;
151   ostream_class->flush = my_output_stream_flush;
152 }
153
154 /* ---------------------------------------------------------------------------------------------------- */
155
156 typedef struct {
157     GError *error;
158     gchar *guid;
159     gboolean flushed;
160
161     GIOStream *client_stream;
162     GInputStream *client_istream;
163     GOutputStream *client_ostream;
164     GOutputStream *client_real_ostream;
165     GDBusConnection *client_conn;
166
167     GIOStream *server_stream;
168     GInputStream *server_istream;
169     GOutputStream *server_ostream;
170     GDBusConnection *server_conn;
171 } Fixture;
172
173 static void
174 setup_client_cb (GObject      *source,
175                  GAsyncResult *res,
176                  gpointer      user_data)
177 {
178   Fixture *f = user_data;
179
180   f->client_conn = g_dbus_connection_new_finish (res, &f->error);
181   g_assert_no_error (f->error);
182   g_assert (G_IS_DBUS_CONNECTION (f->client_conn));
183   g_assert (f->client_conn == G_DBUS_CONNECTION (source));
184 }
185
186 static void
187 setup_server_cb (GObject      *source,
188                  GAsyncResult *res,
189                  gpointer      user_data)
190 {
191   Fixture *f = user_data;
192
193   f->server_conn = g_dbus_connection_new_finish (res, &f->error);
194   g_assert_no_error (f->error);
195   g_assert (G_IS_DBUS_CONNECTION (f->server_conn));
196   g_assert (f->server_conn == G_DBUS_CONNECTION (source));
197 }
198
199 static void
200 setup (Fixture       *f,
201        gconstpointer  test_data G_GNUC_UNUSED)
202 {
203   gboolean ok;
204
205   f->guid = g_dbus_generate_guid ();
206
207   ok = test_pipe (&f->server_istream, &f->client_real_ostream, &f->error);
208   g_assert_no_error (f->error);
209   g_assert (G_IS_OUTPUT_STREAM (f->client_real_ostream));
210   g_assert (G_IS_INPUT_STREAM (f->server_istream));
211   g_assert (ok);
212
213   f->client_ostream = g_object_new (MY_TYPE_OUTPUT_STREAM,
214                                     "base-stream", f->client_real_ostream,
215                                     "close-base-stream", TRUE,
216                                     NULL);
217   g_assert (G_IS_OUTPUT_STREAM (f->client_ostream));
218
219   ok = test_pipe (&f->client_istream, &f->server_ostream, &f->error);
220   g_assert_no_error (f->error);
221   g_assert (G_IS_OUTPUT_STREAM (f->server_ostream));
222   g_assert (G_IS_INPUT_STREAM (f->client_istream));
223   g_assert (ok);
224
225   f->client_stream = test_io_stream_new (f->client_istream, f->client_ostream);
226   f->server_stream = test_io_stream_new (f->server_istream, f->server_ostream);
227
228   g_dbus_connection_new (f->client_stream, NULL,
229                          G_DBUS_CONNECTION_FLAGS_AUTHENTICATION_CLIENT,
230                          NULL, NULL, setup_client_cb, f);
231   g_dbus_connection_new (f->server_stream, f->guid,
232                          G_DBUS_CONNECTION_FLAGS_AUTHENTICATION_SERVER,
233                          NULL, NULL, setup_server_cb, f);
234
235   while (f->client_conn == NULL || f->server_conn == NULL)
236     g_main_context_iteration (NULL, TRUE);
237 }
238
239 static void
240 flush_cb (GObject      *source,
241           GAsyncResult *res,
242           gpointer      user_data)
243 {
244   Fixture *f = user_data;
245   gboolean ok;
246
247   g_assert (G_IS_DBUS_CONNECTION (source));
248   g_assert (G_IS_DBUS_CONNECTION (f->client_conn));
249   g_assert_cmpuint ((guintptr) f->client_conn, ==, (guintptr) G_DBUS_CONNECTION (source));
250
251   ok = g_dbus_connection_flush_finish (f->client_conn, res, &f->error);
252   g_assert_no_error (f->error);
253   g_assert (ok);
254
255   f->flushed = TRUE;
256 }
257
258 static void
259 test_flush_busy (Fixture       *f,
260                  gconstpointer  test_data G_GNUC_UNUSED)
261 {
262   gint initial, started;
263   gboolean ok;
264
265   initial = my_output_stream_get_bytes_started (f->client_ostream);
266   /* make sure the actual write will block */
267   G_LOCK (write);
268
269   ok = g_dbus_connection_emit_signal (f->client_conn, NULL, "/",
270                                       "com.example.Foo", "SomeSignal", NULL,
271                                       &f->error);
272   g_assert_no_error (f->error);
273   g_assert (ok);
274
275   /* wait for at least part of the message to have started writing -
276    * the write will block indefinitely in the worker thread
277    */
278   do {
279     started = my_output_stream_get_bytes_started (f->client_ostream);
280     g_thread_yield ();
281   } while (initial >= started);
282
283   /* we haven't flushed anything */
284   g_assert_cmpint (my_output_stream_get_bytes_flushed (f->client_ostream),
285                    <=, initial);
286
287   /* start to flush: it can't happen til the write finishes */
288   g_dbus_connection_flush (f->client_conn, NULL, flush_cb, f);
289
290   /* we still haven't actually flushed anything */
291   g_assert_cmpint (my_output_stream_get_bytes_flushed (f->client_ostream),
292                    <=, initial);
293
294   /* let the write finish */
295   G_UNLOCK (write);
296
297   /* wait for the flush to happen */
298   while (!f->flushed)
299     g_main_context_iteration (NULL, TRUE);
300
301   /* now we have flushed at least what we'd written - but before fixing
302    * GNOME#662395 this assertion would fail
303    */
304   g_assert_cmpint (my_output_stream_get_bytes_flushed (f->client_ostream),
305                    >=, started);
306 }
307
308 static void
309 test_flush_idle (Fixture       *f,
310                  gconstpointer  test_data G_GNUC_UNUSED)
311 {
312   gint initial, finished;
313   gboolean ok;
314
315   initial = my_output_stream_get_bytes_finished (f->client_ostream);
316
317   ok = g_dbus_connection_emit_signal (f->client_conn, NULL, "/",
318                                       "com.example.Foo", "SomeSignal", NULL,
319                                       &f->error);
320   g_assert_no_error (f->error);
321   g_assert (ok);
322
323   /* wait for at least part of the message to have been written */
324   do {
325     finished = my_output_stream_get_bytes_finished (f->client_ostream);
326     g_thread_yield ();
327   } while (initial >= finished);
328
329   /* we haven't flushed anything */
330   g_assert_cmpint (my_output_stream_get_bytes_flushed (f->client_ostream),
331                    <=, initial);
332
333   /* flush with fully-written, but unflushed, messages */
334   ok = g_dbus_connection_flush_sync (f->client_conn, NULL, &f->error);
335
336   /* now we have flushed at least what we'd written - but before fixing
337    * GNOME#662395 this assertion would fail
338    */
339   g_assert_cmpint (my_output_stream_get_bytes_flushed (f->client_ostream),
340                    >=, finished);
341 }
342
343 static void
344 teardown (Fixture       *f,
345           gconstpointer  test_data G_GNUC_UNUSED)
346 {
347   g_clear_error (&f->error);
348
349   g_clear_object (&f->client_stream);
350   g_clear_object (&f->client_istream);
351   g_clear_object (&f->client_ostream);
352   g_clear_object (&f->client_real_ostream);
353   g_clear_object (&f->client_conn);
354
355   g_clear_object (&f->server_stream);
356   g_clear_object (&f->server_istream);
357   g_clear_object (&f->server_ostream);
358   g_clear_object (&f->server_conn);
359
360   g_free (f->guid);
361 }
362
363 /* ---------------------------------------------------------------------------------------------------- */
364
365 int
366 main (int   argc,
367       char *argv[])
368 {
369   gint ret;
370
371   g_test_init (&argc, &argv, NULL);
372
373   g_test_add ("/gdbus/connection/flush/busy", Fixture, NULL,
374               setup, test_flush_busy, teardown);
375   g_test_add ("/gdbus/connection/flush/idle", Fixture, NULL,
376               setup, test_flush_idle, teardown);
377
378   ret = g_test_run();
379
380   return ret;
381 }