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
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
|
/*
* Copyright (c) 2018-2020, Andreas Kling <kling@serenityos.org>
* Copyright (c) 2022, the SerenityOS developers.
*
* SPDX-License-Identifier: BSD-2-Clause
*/
#pragma once
#include <AK/ByteBuffer.h>
#include <AK/Try.h>
#include <LibCore/Event.h>
#include <LibCore/EventLoop.h>
#include <LibCore/Notifier.h>
#include <LibCore/Socket.h>
#include <LibCore/Timer.h>
#include <LibIPC/Forward.h>
#include <LibIPC/Message.h>
#include <errno.h>
#include <stdint.h>
#include <stdio.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <unistd.h>
namespace IPC {
// NOTE: This is an abstraction to allow using IPC::Connection without a Core::EventLoop.
// FIXME: It's not particularly nice, think of something nicer.
struct DeferredInvoker {
virtual ~DeferredInvoker() = default;
virtual void schedule(Function<void()>) = 0;
};
class ConnectionBase : public Core::Object {
C_OBJECT_ABSTRACT(ConnectionBase);
public:
virtual ~ConnectionBase() override = default;
void set_fd_passing_socket(NonnullOwnPtr<Core::LocalSocket>);
void set_deferred_invoker(NonnullOwnPtr<DeferredInvoker>);
DeferredInvoker& deferred_invoker() { return *m_deferred_invoker; }
bool is_open() const { return m_socket->is_open(); }
ErrorOr<void> post_message(Message const&);
void shutdown();
virtual void die() { }
Core::LocalSocket& socket() { return *m_socket; }
Core::LocalSocket& fd_passing_socket();
protected:
explicit ConnectionBase(IPC::Stub&, NonnullOwnPtr<Core::LocalSocket>, u32 local_endpoint_magic);
virtual void may_have_become_unresponsive() { }
virtual void did_become_responsive() { }
virtual void try_parse_messages(Vector<u8> const& bytes, size_t& index) = 0;
virtual void shutdown_with_error(Error const&);
OwnPtr<IPC::Message> wait_for_specific_endpoint_message_impl(u32 endpoint_magic, int message_id);
void wait_for_socket_to_become_readable();
ErrorOr<Vector<u8>> read_as_much_as_possible_from_socket_without_blocking();
ErrorOr<void> drain_messages_from_peer();
ErrorOr<void> post_message(MessageBuffer);
void handle_messages();
IPC::Stub& m_local_stub;
NonnullOwnPtr<Core::LocalSocket> m_socket;
OwnPtr<Core::LocalSocket> m_fd_passing_socket;
RefPtr<Core::Timer> m_responsiveness_timer;
Vector<NonnullOwnPtr<Message>> m_unprocessed_messages;
ByteBuffer m_unprocessed_bytes;
u32 m_local_endpoint_magic { 0 };
NonnullOwnPtr<DeferredInvoker> m_deferred_invoker;
};
template<typename LocalEndpoint, typename PeerEndpoint>
class Connection : public ConnectionBase {
public:
Connection(IPC::Stub& local_stub, NonnullOwnPtr<Core::LocalSocket> socket)
: ConnectionBase(local_stub, move(socket), LocalEndpoint::static_magic())
{
m_socket->on_ready_to_read = [this] {
NonnullRefPtr protect = *this;
// FIXME: Do something about errors.
(void)drain_messages_from_peer();
handle_messages();
};
}
template<typename MessageType>
OwnPtr<MessageType> wait_for_specific_message()
{
return wait_for_specific_endpoint_message<MessageType, LocalEndpoint>();
}
template<typename RequestType, typename... Args>
NonnullOwnPtr<typename RequestType::ResponseType> send_sync(Args&&... args)
{
MUST(post_message(RequestType(forward<Args>(args)...)));
auto response = wait_for_specific_endpoint_message<typename RequestType::ResponseType, PeerEndpoint>();
VERIFY(response);
return response.release_nonnull();
}
template<typename RequestType, typename... Args>
OwnPtr<typename RequestType::ResponseType> send_sync_but_allow_failure(Args&&... args)
{
if (post_message(RequestType(forward<Args>(args)...)).is_error())
return nullptr;
return wait_for_specific_endpoint_message<typename RequestType::ResponseType, PeerEndpoint>();
}
protected:
template<typename MessageType, typename Endpoint>
OwnPtr<MessageType> wait_for_specific_endpoint_message()
{
if (auto message = wait_for_specific_endpoint_message_impl(Endpoint::static_magic(), MessageType::static_message_id()))
return message.template release_nonnull<MessageType>();
return {};
}
virtual void try_parse_messages(Vector<u8> const& bytes, size_t& index) override
{
u32 message_size = 0;
for (; index + sizeof(message_size) < bytes.size(); index += message_size) {
memcpy(&message_size, bytes.data() + index, sizeof(message_size));
if (message_size == 0 || bytes.size() - index - sizeof(uint32_t) < message_size)
break;
index += sizeof(message_size);
auto remaining_bytes = ReadonlyBytes { bytes.data() + index, message_size };
auto local_message = LocalEndpoint::decode_message(remaining_bytes, fd_passing_socket());
if (!local_message.is_error()) {
m_unprocessed_messages.append(local_message.release_value());
continue;
}
auto peer_message = PeerEndpoint::decode_message(remaining_bytes, fd_passing_socket());
if (!peer_message.is_error()) {
m_unprocessed_messages.append(peer_message.release_value());
continue;
}
dbgln("Failed to parse a message");
dbgln("Local endpoint error: {}", local_message.error());
dbgln("Peer endpoint error: {}", peer_message.error());
break;
}
}
};
}
template<typename LocalEndpoint, typename PeerEndpoint>
struct AK::Formatter<IPC::Connection<LocalEndpoint, PeerEndpoint>> : Formatter<Core::Object> {
};
|
