Merge pull request #10563 from llyzs/transport_io

Add new transport io layer design and support custom socket.

include/freerdp/transport_io.h

@@ -32,6 +32,23 @@
 extern "C"
 {
 #endif
+	typedef int (*pTransportLayerRead)(void* userContext, void* data, int bytes);
+	typedef int (*pTransportLayerWrite)(void* userContext, const void* data, int bytes);
+	typedef BOOL (*pTransportLayerFkt)(void* userContext);
+	typedef BOOL (*pTransportLayerWait)(void* userContext, BOOL waitWrite, DWORD timeout);
+	typedef HANDLE (*pTransportLayerGetEvent)(void* userContext);
+
+	struct rdp_transport_layer
+	{
+		ALIGN64 void* userContext;
+		ALIGN64 pTransportLayerRead Read;
+		ALIGN64 pTransportLayerWrite Write;
+		ALIGN64 pTransportLayerFkt Close;
+		ALIGN64 pTransportLayerWait Wait;
+		ALIGN64 pTransportLayerGetEvent GetEvent;
+		UINT64 reserved[64 - 6]; /* Reserve some space for ABI compatibility */
+	};
+	typedef struct rdp_transport_layer rdpTransportLayer;
 
 	typedef int (*pTCPConnect)(rdpContext* context, rdpSettings* settings, const char* hostname,
 	                           int port, DWORD timeout);
@@ -42,6 +59,10 @@ extern "C"
 	typedef BOOL (*pTransportGetPublicKey)(rdpTransport* transport, const BYTE** data,
 	                                       DWORD* length);
 	typedef BOOL (*pTransportSetBlockingMode)(rdpTransport* transport, BOOL blocking);
+	typedef rdpTransportLayer* (*pTransportConnectLayer)(rdpTransport* transport,
+	                                                     const char* hostname, int port,
+	                                                     DWORD timeout);
+	typedef BOOL (*pTransportAttachLayer)(rdpTransport* transport, rdpTransportLayer* layer);
 
 	struct rdp_transport_io
 	{
@@ -52,10 +73,12 @@ extern "C"
 		pTransportFkt TransportDisconnect;
 		pTransportRWFkt ReadPdu;  /* Reads a whole PDU from the transport */
 		pTransportRWFkt WritePdu; /* Writes a whole PDU to the transport */
-		pTransportRead ReadBytes; /* Reads up to a requested amount of bytes from the transport */
+		pTransportRead ReadBytes; /* Reads up to a requested amount of bytes */
 		pTransportGetPublicKey GetPublicKey;
 		pTransportSetBlockingMode SetBlockingMode;
-		UINT64 reserved[54]; /* Reserve some space for ABI compatibility */
+		pTransportConnectLayer ConnectLayer;
+		pTransportAttachLayer AttachLayer;
+		UINT64 reserved[64 - 12]; /* Reserve some space for ABI compatibility */
 	};
 	typedef struct rdp_transport_io rdpTransportIo;
 
@@ -78,6 +101,9 @@ extern "C"
 	FREERDP_API rdpContext* transport_get_context(rdpTransport* transport);
 	FREERDP_API rdpTransport* freerdp_get_transport(rdpContext* context);
 
+	FREERDP_API rdpTransportLayer* transport_layer_new(rdpTransport* transport, size_t contextSize);
+	FREERDP_API void transport_layer_free(rdpTransportLayer* layer);
+
 #ifdef __cplusplus
 }
 #endif

libfreerdp/core/gateway/rdg.c

@@ -1138,16 +1138,7 @@ DWORD rdg_get_event_handles(rdpRdg* rdg, HANDLE* events, DWORD count)
 			return 0;
 	}
 
-	if (!rdg->transferEncoding.isWebsocketTransport && rdg->tlsIn && rdg->tlsIn->bio)
+	/* We just need the read event handle even in non-websocket mode. */
-	{
-		if (events && (nCount < count))
-		{
-			BIO_get_event(rdg->tlsIn->bio, &events[nCount]);
-			nCount++;
-		}
-		else
-			return 0;
-	}
 
 	return nCount;
 }
@@ -1266,11 +1257,12 @@ static BOOL rdg_send_http_request(rdpRdg* rdg, rdpTls* tls, const char* method,
 
 static BOOL rdg_tls_connect(rdpRdg* rdg, rdpTls* tls, const char* peerAddress, int timeout)
 {
-	int sockfd = 0;
 	long status = 0;
-	BIO* socketBio = NULL;
+	BIO* layerBio = NULL;
 	BIO* bufferedBio = NULL;
+	rdpTransportLayer* layer = NULL;
 	rdpSettings* settings = rdg->context->settings;
+	rdpTransport* transport = freerdp_get_transport(rdg->context);
 	const char* peerHostname = settings->GatewayHostname;
 	UINT16 peerPort = (UINT16)settings->GatewayPort;
 	const char* proxyUsername = NULL;
@@ -1281,32 +1273,30 @@ static BOOL rdg_tls_connect(rdpRdg* rdg, rdpTls* tls, const char* peerAddress, i
 	if (settings->GatewayPort > UINT16_MAX)
 		return FALSE;
 
-	sockfd = freerdp_tcp_connect(rdg->context, peerAddress ? peerAddress : peerHostname, peerPort,
+	layer = transport_connect_layer(transport, peerAddress ? peerAddress : peerHostname, peerPort,
 	                                timeout);
 
-	if (sockfd < 0)
+	if (!layer)
 	{
 		return FALSE;
 	}
 
-	socketBio = BIO_new(BIO_s_simple_socket());
+	layerBio = BIO_new(BIO_s_transport_layer());
-
+	if (!layerBio)
-	if (!socketBio)
 	{
-		closesocket((SOCKET)sockfd);
+		transport_layer_free(layer);
 		return FALSE;
 	}
+	BIO_set_data(layerBio, layer);
 
-	BIO_set_fd(socketBio, sockfd, BIO_CLOSE);
 	bufferedBio = BIO_new(BIO_s_buffered_socket());
-
 	if (!bufferedBio)
 	{
-		BIO_free_all(socketBio);
+		BIO_free_all(layerBio);
 		return FALSE;
 	}
 
-	bufferedBio = BIO_push(bufferedBio, socketBio);
+	bufferedBio = BIO_push(bufferedBio, layerBio);
 	status = BIO_set_nonblock(bufferedBio, TRUE);
 
 	if (isProxyConnection)

libfreerdp/core/streamdump.c

@@ -386,6 +386,16 @@ static int stream_dump_replay_transport_tcp_connect(rdpContext* context, rdpSett
 	return 42;
 }
 
+static rdpTransportLayer* stream_dump_replay_transport_connect_layer(rdpTransport* transport,
+                                                                     const char* hostname, int port,
+                                                                     DWORD timeout)
+{
+	WINPR_ASSERT(transport);
+	WINPR_ASSERT(hostname);
+
+	return NULL;
+}
+
 static BOOL stream_dump_replay_transport_tls_connect(rdpTransport* transport)
 {
 	WINPR_ASSERT(transport);
@@ -425,6 +435,7 @@ static BOOL stream_dump_register_read_handlers(rdpContext* context)
 	dump.TCPConnect = stream_dump_replay_transport_tcp_connect;
 	dump.TLSAccept = stream_dump_replay_transport_accept;
 	dump.TLSConnect = stream_dump_replay_transport_tls_connect;
+	dump.ConnectLayer = stream_dump_replay_transport_connect_layer;
 	if (!freerdp_set_io_callbacks(context, &dump))
 		return FALSE;
 	return freerdp_io_callback_set_event(context, TRUE);

libfreerdp/core/tcp.c

@@ -1336,3 +1336,200 @@ int freerdp_tcp_default_connect(rdpContext* context, rdpSettings* settings, cons
 
 	return sockfd;
 }
+
+struct rdp_tcp_layer
+{
+	int sockfd;
+	HANDLE hEvent;
+};
+typedef struct rdp_tcp_layer rdpTcpLayer;
+
+static int freerdp_tcp_layer_read(void* userContext, void* data, int bytes)
+{
+	if (!userContext)
+		return -1;
+	if (!data || !bytes)
+		return 0;
+
+	rdpTcpLayer* tcpLayer = (rdpTcpLayer*)userContext;
+
+	int error = 0;
+	int status = 0;
+
+	WSAResetEvent(tcpLayer->hEvent);
+	status = _recv(tcpLayer->sockfd, data, bytes, 0);
+
+	if (status > 0)
+		return status;
+
+	if (status == 0)
+		return -1; /* socket closed */
+
+	error = WSAGetLastError();
+
+	if ((error == WSAEWOULDBLOCK) || (error == WSAEINTR) || (error == WSAEINPROGRESS) ||
+	    (error == WSAEALREADY))
+	{
+		status = 0;
+	}
+	else
+	{
+		status = -1;
+	}
+
+	return status;
+}
+
+static int freerdp_tcp_layer_write(void* userContext, const void* data, int bytes)
+{
+	if (!userContext)
+		return -1;
+	if (!data || !bytes)
+		return 0;
+
+	rdpTcpLayer* tcpLayer = (rdpTcpLayer*)userContext;
+
+	int error = 0;
+	int status = 0;
+
+	status = _send(tcpLayer->sockfd, data, bytes, 0);
+
+	if (status <= 0)
+	{
+		error = WSAGetLastError();
+
+		if ((error == WSAEWOULDBLOCK) || (error == WSAEINTR) || (error == WSAEINPROGRESS) ||
+		    (error == WSAEALREADY))
+		{
+			status = 0;
+		}
+		else
+		{
+			status = -1;
+		}
+	}
+
+	return status;
+}
+
+static BOOL freerdp_tcp_layer_close(void* userContext)
+{
+	if (!userContext)
+		return FALSE;
+
+	rdpTcpLayer* tcpLayer = (rdpTcpLayer*)userContext;
+
+	if (tcpLayer->sockfd >= 0)
+		closesocket(tcpLayer->sockfd);
+	if (tcpLayer->hEvent)
+		CloseHandle(tcpLayer->hEvent);
+
+	return TRUE;
+}
+
+static BOOL freerdp_tcp_layer_wait(void* userContext, BOOL waitWrite, DWORD timeout)
+{
+	if (!userContext)
+		return FALSE;
+
+	rdpTcpLayer* tcpLayer = (rdpTcpLayer*)userContext;
+
+	int status = -1;
+	int sockfd = tcpLayer->sockfd;
+#ifdef WINPR_HAVE_POLL_H
+	struct pollfd pollset = { 0 };
+	pollset.fd = sockfd;
+	pollset.events = waitWrite ? POLLOUT : POLLIN;
+
+	do
+	{
+		status = poll(&pollset, 1, (int)timeout);
+	} while ((status < 0) && (errno == EINTR));
+
+#else
+	fd_set rset;
+	struct timeval tv;
+	FD_ZERO(&rset);
+	FD_SET(sockfd, &rset);
+
+	if (timeout)
+	{
+		tv.tv_sec = timeout / 1000;
+		tv.tv_usec = (timeout % 1000) * 1000;
+	}
+
+	do
+	{
+		if (waitWrite)
+			status = select(sockfd + 1, NULL, &rset, NULL, timeout ? &tv : NULL);
+		else
+			status = select(sockfd + 1, &rset, NULL, NULL, timeout ? &tv : NULL);
+	} while ((status < 0) && (errno == EINTR));
+
+#endif
+
+	return status != 0;
+}
+
+static HANDLE freerdp_tcp_layer_get_event(void* userContext)
+{
+	if (!userContext)
+		return NULL;
+
+	rdpTcpLayer* tcpLayer = (rdpTcpLayer*)userContext;
+
+	return tcpLayer->hEvent;
+}
+
+rdpTransportLayer* freerdp_tcp_connect_layer(rdpContext* context, const char* hostname, int port,
+                                             DWORD timeout)
+{
+	WINPR_ASSERT(context);
+
+	const rdpSettings* settings = context->settings;
+	WINPR_ASSERT(settings);
+
+	rdpTransportLayer* layer = NULL;
+	rdpTcpLayer* tcpLayer = NULL;
+
+	int sockfd = freerdp_tcp_connect(context, hostname, port, timeout);
+	if (sockfd < 0)
+		goto fail;
+	if (!freerdp_tcp_set_keep_alive_mode(settings, sockfd))
+		goto fail;
+
+	layer = transport_layer_new(freerdp_get_transport(context), sizeof(rdpTcpLayer));
+	if (!layer)
+		goto fail;
+
+	layer->Read = freerdp_tcp_layer_read;
+	layer->Write = freerdp_tcp_layer_write;
+	layer->Close = freerdp_tcp_layer_close;
+	layer->Wait = freerdp_tcp_layer_wait;
+	layer->GetEvent = freerdp_tcp_layer_get_event;
+
+	tcpLayer = (rdpTcpLayer*)layer->userContext;
+	WINPR_ASSERT(tcpLayer);
+
+	tcpLayer->sockfd = -1;
+	tcpLayer->hEvent = WSACreateEvent();
+	if (!tcpLayer->hEvent)
+		goto fail;
+
+	/* WSAEventSelect automatically sets the socket in non-blocking mode */
+	if (WSAEventSelect(sockfd, tcpLayer->hEvent, FD_READ | FD_ACCEPT | FD_CLOSE))
+	{
+		WLog_ERR(TAG, "WSAEventSelect returned 0x%08X", WSAGetLastError());
+		goto fail;
+	}
+
+	tcpLayer->sockfd = sockfd;
+
+	return layer;
+
+fail:
+	if (sockfd >= 0)
+		closesocket(sockfd);
+	transport_layer_free(layer);
+	return NULL;
+}

libfreerdp/core/tcp.h

@@ -27,6 +27,7 @@
 #include <freerdp/settings.h>
 #include <freerdp/freerdp.h>
 #include <freerdp/api.h>
+#include <freerdp/transport_io.h>
 
 #include <winpr/crt.h>
 #include <winpr/synch.h>
@@ -102,6 +103,9 @@ FREERDP_LOCAL int freerdp_tcp_connect(rdpContext* context, const char* hostname,
 FREERDP_LOCAL int freerdp_tcp_default_connect(rdpContext* context, rdpSettings* settings,
                                               const char* hostname, int port, DWORD timeout);
 
+FREERDP_LOCAL rdpTransportLayer*
+freerdp_tcp_connect_layer(rdpContext* context, const char* hostname, int port, DWORD timeout);
+
 FREERDP_LOCAL char* freerdp_tcp_get_peer_address(SOCKET sockfd);
 
 FREERDP_LOCAL struct addrinfo* freerdp_tcp_resolve_host(const char* hostname, int port,

libfreerdp/core/transport.c

@@ -471,7 +471,6 @@ BOOL transport_connect_aad(rdpTransport* transport)
 
 BOOL transport_connect(rdpTransport* transport, const char* hostname, UINT16 port, DWORD timeout)
 {
-	int sockfd = 0;
 	BOOL status = FALSE;
 	rdpSettings* settings = NULL;
 	rdpContext* context = transport_get_context(transport);
@@ -570,16 +569,20 @@ BOOL transport_connect(rdpTransport* transport, const char* hostname, UINT16 por
 		BOOL isProxyConnection =
 		    proxy_prepare(settings, &proxyHostname, &peerPort, &proxyUsername, &proxyPassword);
 
+		rdpTransportLayer* layer = NULL;
 		if (isProxyConnection)
-			sockfd = transport_tcp_connect(transport, proxyHostname, peerPort, timeout);
+			layer = transport_connect_layer(transport, proxyHostname, peerPort, timeout);
 		else
-			sockfd = transport_tcp_connect(transport, hostname, port, timeout);
+			layer = transport_connect_layer(transport, hostname, port, timeout);
 
-		if (sockfd < 0)
+		if (!layer)
 			return FALSE;
 
-		if (!transport_attach(transport, sockfd))
+		if (!transport_attach_layer(transport, layer))
+		{
+			transport_layer_free(layer);
 			return FALSE;
+		}
 
 		if (isProxyConnection)
 		{
@@ -1477,6 +1480,59 @@ static BOOL transport_default_set_blocking_mode(rdpTransport* transport, BOOL bl
 	return TRUE;
 }
 
+rdpTransportLayer* transport_connect_layer(rdpTransport* transport, const char* hostname, int port,
+                                           DWORD timeout)
+{
+	WINPR_ASSERT(transport);
+
+	return IFCALLRESULT(NULL, transport->io.ConnectLayer, transport, hostname, port, timeout);
+}
+
+rdpTransportLayer* transport_default_connect_layer(rdpTransport* transport, const char* hostname,
+                                                   int port, DWORD timeout)
+{
+	rdpContext* context = transport_get_context(transport);
+	WINPR_ASSERT(context);
+
+	return freerdp_tcp_connect_layer(context, hostname, port, timeout);
+}
+
+BOOL transport_attach_layer(rdpTransport* transport, rdpTransportLayer* layer)
+{
+	WINPR_ASSERT(transport);
+	WINPR_ASSERT(layer);
+
+	return IFCALLRESULT(FALSE, transport->io.AttachLayer, transport, layer);
+}
+
+BOOL transport_default_attach_layer(rdpTransport* transport, rdpTransportLayer* layer)
+{
+	BIO* layerBio = BIO_new(BIO_s_transport_layer());
+	if (!layerBio)
+		goto fail;
+
+	BIO* bufferedBio = BIO_new(BIO_s_buffered_socket());
+	if (!bufferedBio)
+		goto fail;
+
+	bufferedBio = BIO_push(bufferedBio, layerBio);
+	if (!bufferedBio)
+		goto fail;
+
+	/* BIO takes over the layer reference at this point. */
+	BIO_set_data(layerBio, layer);
+
+	transport->frontBio = bufferedBio;
+
+	return TRUE;
+
+fail:
+	if (layerBio)
+		BIO_free_all(layerBio);
+
+	return FALSE;
+}
+
 void transport_set_gateway_enabled(rdpTransport* transport, BOOL GatewayEnabled)
 {
 	WINPR_ASSERT(transport);
@@ -1574,6 +1630,8 @@ rdpTransport* transport_new(rdpContext* context)
 	transport->io.ReadBytes = transport_read_layer;
 	transport->io.GetPublicKey = transport_default_get_public_key;
 	transport->io.SetBlockingMode = transport_default_set_blocking_mode;
+	transport->io.ConnectLayer = transport_default_connect_layer;
+	transport->io.AttachLayer = transport_default_attach_layer;
 
 	transport->context = context;
 	transport->ReceivePool = StreamPool_New(TRUE, BUFFER_SIZE);
@@ -1809,3 +1867,206 @@ void transport_set_early_user_auth_mode(rdpTransport* transport, BOOL EUAMode)
 	transport->earlyUserAuth = EUAMode;
 	WLog_Print(transport->log, WLOG_DEBUG, "Early User Auth Mode: %s", EUAMode ? "on" : "off");
 }
+
+rdpTransportLayer* transport_layer_new(rdpTransport* transport, size_t contextSize)
+{
+	rdpTransportLayer* layer = (rdpTransportLayer*)calloc(1, sizeof(rdpTransportLayer));
+	if (!layer)
+		return NULL;
+
+	if (contextSize)
+	{
+		layer->userContext = calloc(1, contextSize);
+		if (!layer->userContext)
+		{
+			free(layer);
+			return NULL;
+		}
+	}
+
+	return layer;
+}
+
+void transport_layer_free(rdpTransportLayer* layer)
+{
+	if (!layer)
+		return;
+
+	IFCALL(layer->Close, layer->userContext);
+	free(layer->userContext);
+	free(layer);
+}
+
+static int transport_layer_bio_write(BIO* bio, const char* buf, int size)
+{
+	if (!buf || !size)
+		return 0;
+	if (size < 0)
+		return -1;
+
+	WINPR_ASSERT(bio);
+
+	rdpTransportLayer* layer = (rdpTransportLayer*)BIO_get_data(bio);
+	if (!layer)
+		return -1;
+
+	BIO_clear_flags(bio, BIO_FLAGS_WRITE | BIO_FLAGS_SHOULD_RETRY);
+
+	int status = IFCALLRESULT(-1, layer->Write, layer->userContext, buf, size);
+
+	if (status >= 0 && status < size)
+		BIO_set_flags(bio, (BIO_FLAGS_WRITE | BIO_FLAGS_SHOULD_RETRY));
+
+	return status;
+}
+
+static int transport_layer_bio_read(BIO* bio, char* buf, int size)
+{
+	if (!buf || !size)
+		return 0;
+	if (size < 0)
+		return -1;
+
+	WINPR_ASSERT(bio);
+
+	rdpTransportLayer* layer = (rdpTransportLayer*)BIO_get_data(bio);
+	if (!layer)
+		return -1;
+
+	BIO_clear_flags(bio, BIO_FLAGS_READ | BIO_FLAGS_SHOULD_RETRY);
+
+	int status = IFCALLRESULT(-1, layer->Read, layer->userContext, buf, size);
+
+	if (status == 0)
+		BIO_set_flags(bio, (BIO_FLAGS_READ | BIO_FLAGS_SHOULD_RETRY));
+
+	return status;
+}
+
+static int transport_layer_bio_puts(BIO* bio, const char* str)
+{
+	return 1;
+}
+
+static int transport_layer_bio_gets(BIO* bio, char* str, int size)
+{
+	return 1;
+}
+
+static long transport_layer_bio_ctrl(BIO* bio, int cmd, long arg1, void* arg2)
+{
+	WINPR_ASSERT(bio);
+
+	rdpTransportLayer* layer = (rdpTransportLayer*)BIO_get_data(bio);
+	if (!layer)
+		return -1;
+
+	int status = -1;
+	switch (cmd)
+	{
+		case BIO_C_GET_EVENT:
+			*((HANDLE*)arg2) = IFCALLRESULT(NULL, layer->GetEvent, layer->userContext);
+			status = 1;
+			break;
+
+		case BIO_C_SET_NONBLOCK:
+			status = 1;
+			break;
+
+		case BIO_C_WAIT_READ:
+		{
+			int timeout = (int)arg1;
+			BOOL r = IFCALLRESULT(FALSE, layer->Wait, layer->userContext, FALSE, timeout);
+			/* Convert timeout to error return */
+			if (!r)
+			{
+				errno = ETIMEDOUT;
+				status = 0;
+			}
+			else
+				status = 1;
+			break;
+		}
+
+		case BIO_C_WAIT_WRITE:
+		{
+			int timeout = (int)arg1;
+			BOOL r = IFCALLRESULT(FALSE, layer->Wait, layer->userContext, TRUE, timeout);
+			/* Convert timeout to error return */
+			if (!r)
+			{
+				errno = ETIMEDOUT;
+				status = 0;
+			}
+			else
+				status = 1;
+			break;
+		}
+
+		case BIO_CTRL_GET_CLOSE:
+			status = BIO_get_shutdown(bio);
+			break;
+
+		case BIO_CTRL_SET_CLOSE:
+			BIO_set_shutdown(bio, (int)arg1);
+			status = 1;
+			break;
+
+		case BIO_CTRL_FLUSH:
+		case BIO_CTRL_DUP:
+			status = 1;
+			break;
+
+		default:
+			status = 0;
+			break;
+	}
+
+	return status;
+}
+
+static int transport_layer_bio_new(BIO* bio)
+{
+	WINPR_ASSERT(bio);
+
+	BIO_set_flags(bio, BIO_FLAGS_SHOULD_RETRY);
+	BIO_set_init(bio, 1);
+	return 1;
+}
+
+static int transport_layer_bio_free(BIO* bio)
+{
+	if (!bio)
+		return 0;
+
+	rdpTransportLayer* layer = (rdpTransportLayer*)BIO_get_data(bio);
+	if (layer)
+		transport_layer_free(layer);
+
+	BIO_set_data(bio, NULL);
+	BIO_set_init(bio, 0);
+	BIO_set_flags(bio, 0);
+
+	return 1;
+}
+
+BIO_METHOD* BIO_s_transport_layer(void)
+{
+	static BIO_METHOD* bio_methods = NULL;
+
+	if (bio_methods == NULL)
+	{
+		if (!(bio_methods = BIO_meth_new(BIO_TYPE_SIMPLE, "TransportLayer")))
+			return NULL;
+
+		BIO_meth_set_write(bio_methods, transport_layer_bio_write);
+		BIO_meth_set_read(bio_methods, transport_layer_bio_read);
+		BIO_meth_set_puts(bio_methods, transport_layer_bio_puts);
+		BIO_meth_set_gets(bio_methods, transport_layer_bio_gets);
+		BIO_meth_set_ctrl(bio_methods, transport_layer_bio_ctrl);
+		BIO_meth_set_create(bio_methods, transport_layer_bio_new);
+		BIO_meth_set_destroy(bio_methods, transport_layer_bio_free);
+	}
+
+	return bio_methods;
+}

libfreerdp/core/transport.h

@@ -74,6 +74,23 @@ FREERDP_LOCAL BOOL transport_connect_childsession(rdpTransport* transport);
  * \return \b TRUE in case of success, \b FALSE otherwise.
  */
 FREERDP_LOCAL BOOL transport_attach(rdpTransport* transport, int sockfd);
+
+/**! \brief Attach a transport layer
+ *
+ * The ownership of the transport layer provided by \b layer is taken if and only if the function is
+ * successful. In such a case the caller must no longer free or otherwise use the layer. If the
+ * function fails it is up to the caller to free the layer.
+ *
+ * The implementation can be overridden by
+ * transport_set_io_callbacks(rdpTransportIo::AttachLayer)
+ *
+ * \param transport The transport instance to attach the socket to
+ * \param layer The layer to attach to the transport
+ *
+ * \return \b TRUE in case of success, \b FALSE otherwise.
+ */
+FREERDP_LOCAL BOOL transport_attach_layer(rdpTransport* transport, rdpTransportLayer* layer);
+
 FREERDP_LOCAL BOOL transport_disconnect(rdpTransport* transport);
 FREERDP_LOCAL BOOL transport_connect_rdp(rdpTransport* transport);
 FREERDP_LOCAL BOOL transport_connect_tls(rdpTransport* transport);
@@ -108,9 +125,6 @@ FREERDP_LOCAL void transport_set_aad_mode(rdpTransport* transport, BOOL AadMode)
 FREERDP_LOCAL BOOL transport_is_write_blocked(rdpTransport* transport);
 FREERDP_LOCAL int transport_drain_output_buffer(rdpTransport* transport);
 
-FREERDP_LOCAL wStream* transport_receive_pool_take(rdpTransport* transport);
-FREERDP_LOCAL int transport_receive_pool_return(rdpTransport* transport, wStream* pdu);
-
 FREERDP_LOCAL BOOL transport_io_callback_set_event(rdpTransport* transport, BOOL set);
 
 FREERDP_LOCAL const rdpTransportIo* transport_get_io_callbacks(rdpTransport* transport);
@@ -146,6 +160,9 @@ FREERDP_LOCAL BOOL transport_set_recv_callbacks(rdpTransport* transport, Transpo
 FREERDP_LOCAL int transport_tcp_connect(rdpTransport* transport, const char* hostname, int port,
                                         DWORD timeout);
 
+FREERDP_LOCAL rdpTransportLayer*
+transport_connect_layer(rdpTransport* transport, const char* hostname, int port, DWORD timeout);
+
 FREERDP_LOCAL void transport_free(rdpTransport* transport);
 
 WINPR_ATTR_MALLOC(transport_free, 1)
@@ -153,4 +170,6 @@ FREERDP_LOCAL rdpTransport* transport_new(rdpContext* context);
 
 FREERDP_LOCAL void transport_set_early_user_auth_mode(rdpTransport* transport, BOOL EUAMode);
 
+FREERDP_LOCAL BIO_METHOD* BIO_s_transport_layer(void);
+
 #endif /* FREERDP_LIB_CORE_TRANSPORT_H */