server: Added support for flushing client pipe.

dlls/kernel32/tests/pipe.c

@@ -2853,6 +2853,9 @@ static void test_overlapped_transport(BOOL msg_mode, BOOL msg_read_mode)
     trace("testing %s, %s server->client writes...\n",
           msg_mode ? "message mode" : "byte mode", msg_read_mode ? "message read" : "byte read");
     test_blocking_rw(server, client, 5000, msg_mode, msg_read_mode);
+    trace("testing %s, %s client->server writes...\n",
+          msg_mode ? "message mode" : "byte mode", msg_read_mode ? "message read" : "byte read");
+    test_blocking_rw(client, server, 6000, msg_mode, msg_read_mode);
 
     CloseHandle(client);
     CloseHandle(server);
@@ -2869,18 +2872,22 @@ static void test_overlapped_transport(BOOL msg_mode, BOOL msg_read_mode)
     /* close server with pending writes */
     create_overlapped_pipe(create_flags, &client, &server);
     overlapped_write_async(client, buf, 7000, &overlapped);
+    flush = test_flush_async(client, ERROR_BROKEN_PIPE);
     CloseHandle(server);
     test_overlapped_failure(client, &overlapped, ERROR_BROKEN_PIPE);
+    test_flush_done(flush);
     CloseHandle(client);
 
     /* disconnect with pending writes */
     create_overlapped_pipe(create_flags, &client, &server);
     overlapped_write_async(client, buf, 7000, &overlapped);
     overlapped_write_async(server, buf, 7000, &overlapped2);
+    flush = test_flush_async(client, ERROR_PIPE_NOT_CONNECTED);
     res = DisconnectNamedPipe(server);
     ok(res, "DisconnectNamedPipe failed: %u\n", GetLastError());
     test_overlapped_failure(client, &overlapped, ERROR_PIPE_NOT_CONNECTED);
     test_overlapped_failure(client, &overlapped2, ERROR_PIPE_NOT_CONNECTED);
+    test_flush_done(flush);
     CloseHandle(server);
     CloseHandle(client);
 }

server/named_pipe.c

@@ -656,6 +656,9 @@ static obj_handle_t pipe_end_flush( struct pipe_end *pipe_end, struct async *asy
 {
     obj_handle_t handle = 0;
 
+    if (use_server_io( pipe_end ) && (!pipe_end->connection || list_empty( &pipe_end->connection->message_queue )))
+        return 0;
+
     if (!fd_queue_async( pipe_end->fd, async, ASYNC_TYPE_WAIT )) return 0;
 
     if (!blocking || (handle = alloc_handle( current->process, async, SYNCHRONIZE, 0 )))
@@ -682,8 +685,9 @@ static obj_handle_t pipe_server_flush( struct fd *fd, struct async *async, int b
 
 static obj_handle_t pipe_client_flush( struct fd *fd, struct async *async, int blocking )
 {
-    /* FIXME: what do we have to do for this? */
-    return 0;
+    struct pipe_end *pipe_end = get_fd_user( fd );
+    /* FIXME: Support byte mode. */
+    return use_server_io( pipe_end ) ? pipe_end_flush( pipe_end, async, blocking ) : 0;
 }
 
 static void message_queue_read( struct pipe_end *pipe_end, struct iosb *iosb )