aboutsummaryrefslogtreecommitdiffstats
path: root/src/event/BufferedSocket.cxx
diff options
context:
space:
mode:
Diffstat (limited to '')
-rw-r--r--src/event/BufferedSocket.cxx256
1 files changed, 256 insertions, 0 deletions
diff --git a/src/event/BufferedSocket.cxx b/src/event/BufferedSocket.cxx
new file mode 100644
index 000000000..f84fe808b
--- /dev/null
+++ b/src/event/BufferedSocket.cxx
@@ -0,0 +1,256 @@
+/*
+ * Copyright (C) 2003-2013 The Music Player Daemon Project
+ * http://www.musicpd.org
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; either version 2 of the License, 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 Street, Fifth Floor, Boston, MA 02110-1301 USA.
+ */
+
+#include "config.h"
+#include "BufferedSocket.hxx"
+#include "SocketError.hxx"
+#include "util/fifo_buffer.h"
+
+#include <assert.h>
+#include <stdint.h>
+#include <string.h>
+
+#ifndef WIN32
+#include <sys/types.h>
+#include <sys/socket.h>
+#endif
+
+BufferedSocket::~BufferedSocket()
+{
+ if (input != nullptr)
+ fifo_buffer_free(input);
+}
+
+BufferedSocket::ssize_t
+BufferedSocket::DirectWrite(const void *data, size_t length)
+{
+ int flags = 0;
+#ifdef MSG_NOSIGNAL
+ flags |= MSG_NOSIGNAL;
+#endif
+#ifdef MSG_DONTWAIT
+ flags |= MSG_DONTWAIT;
+#endif
+
+ const auto nbytes = send(Get(), (const char *)data, length, flags);
+ if (gcc_unlikely(nbytes < 0)) {
+ const auto code = GetSocketError();
+ if (IsSocketErrorAgain(code))
+ return 0;
+
+ Cancel();
+
+ if (IsSocketErrorClosed(code))
+ OnSocketClosed();
+ else
+ OnSocketError(NewSocketError(code));
+ }
+
+ return nbytes;
+}
+
+ssize_t
+BufferedSocket::DirectRead(void *data, size_t length)
+{
+ int flags = 0;
+#ifdef MSG_DONTWAIT
+ flags |= MSG_DONTWAIT;
+#endif
+
+ const auto nbytes = recv(Get(), (char *)data, length, flags);
+ if (gcc_likely(nbytes > 0))
+ return nbytes;
+
+ if (nbytes == 0) {
+ OnSocketClosed();
+ return -1;
+ }
+
+ const auto code = GetSocketError();
+ if (IsSocketErrorAgain(code))
+ return 0;
+
+ if (IsSocketErrorClosed(code))
+ OnSocketClosed();
+ else
+ OnSocketError(NewSocketError(code));
+ return -1;
+}
+
+bool
+BufferedSocket::WriteFromBuffer()
+{
+ assert(IsDefined());
+
+ size_t length;
+ const void *data = output.Read(&length);
+ if (data == nullptr) {
+ CancelWrite();
+ return true;
+ }
+
+ auto nbytes = DirectWrite(data, length);
+ if (gcc_unlikely(nbytes <= 0))
+ return nbytes == 0;
+
+ output.Consume(nbytes);
+
+ if (output.IsEmpty())
+ CancelWrite();
+
+ return true;
+}
+
+bool
+BufferedSocket::ReadToBuffer()
+{
+ assert(IsDefined());
+
+ if (input == nullptr)
+ input = fifo_buffer_new(8192);
+
+ size_t length;
+ void *buffer = fifo_buffer_write(input, &length);
+ assert(buffer != nullptr);
+
+ const auto nbytes = DirectRead(buffer, length);
+ if (nbytes > 0)
+ fifo_buffer_append(input, nbytes);
+
+ return nbytes >= 0;
+}
+
+bool
+BufferedSocket::Write(const void *data, size_t length)
+{
+ assert(IsDefined());
+
+#if 0
+ /* TODO: disabled because this would add overhead on some callers (the ones that often), but it may be useful */
+
+ if (output.IsEmpty()) {
+ /* try to write it directly first */
+ const auto nbytes = DirectWrite(data, length);
+ if (gcc_likely(nbytes > 0)) {
+ data = (const uint8_t *)data + nbytes;
+ length -= nbytes;
+ if (length == 0)
+ return true;
+ } else if (nbytes < 0)
+ return false;
+ }
+#endif
+
+ if (!output.Append(data, length)) {
+ // TODO
+ OnSocketError(g_error_new_literal(g_quark_from_static_string("buffered_socket"),
+ 0, "Output buffer is full"));
+ return false;
+ }
+
+ ScheduleWrite();
+ return true;
+}
+
+bool
+BufferedSocket::ResumeInput()
+{
+ assert(IsDefined());
+
+ if (input == nullptr) {
+ ScheduleRead();
+ return true;
+ }
+
+ while (true) {
+ size_t length;
+ const void *data = fifo_buffer_read(input, &length);
+ if (data == nullptr) {
+ ScheduleRead();
+ return true;
+ }
+
+ const auto result = OnSocketInput(data, length);
+ switch (result) {
+ case InputResult::MORE:
+ if (fifo_buffer_is_full(input)) {
+ // TODO
+ OnSocketError(g_error_new_literal(g_quark_from_static_string("buffered_socket"),
+ 0, "Input buffer is full"));
+ return false;
+ }
+
+ ScheduleRead();
+ return true;
+
+ case InputResult::PAUSE:
+ CancelRead();
+ return true;
+
+ case InputResult::AGAIN:
+ continue;
+
+ case InputResult::CLOSED:
+ return false;
+ }
+ }
+}
+
+void
+BufferedSocket::ConsumeInput(size_t nbytes)
+{
+ assert(IsDefined());
+
+ fifo_buffer_consume(input, nbytes);
+}
+
+void
+BufferedSocket::OnSocketReady(unsigned flags)
+{
+ assert(IsDefined());
+
+ if (gcc_unlikely(flags & (ERROR|HANGUP))) {
+ OnSocketClosed();
+ return;
+ }
+
+ if (flags & READ) {
+ assert(input == nullptr || !fifo_buffer_is_full(input));
+
+ if (!ReadToBuffer() || !ResumeInput())
+ return;
+
+ if (input == nullptr || !fifo_buffer_is_full(input))
+ ScheduleRead();
+
+ /* just in case the OnSocketInput() method has added
+ data to the output buffer: try to send it now
+ instead of waiting for the next event loop
+ iteration */
+ if (!output.IsEmpty())
+ flags |= WRITE;
+ }
+
+ if (flags & WRITE) {
+ assert(!output.IsEmpty());
+
+ if (!WriteFromBuffer())
+ return;
+ }
+}