Add a GIO version of accept_async

Cargo.toml

@@ -140,6 +140,10 @@ required-features = ["async-std-runtime"]
 name = "gio-echo"
 required-features = ["gio-runtime"]
 
+[[example]]
+name = "gio-echo-server"
+required-features = ["gio-runtime"]
+
 [[example]]
 name = "tokio-echo"
 required-features = ["tokio-runtime"]

examples/gio-echo-server.rs

@@ -0,0 +1,64 @@
+use std::{env, net::SocketAddr};
+
+use async_tungstenite::{
+    gio::accept_async,
+    tungstenite::{Error, Result},
+};
+use futures::prelude::*;
+use gio::{
+    prelude::*, InetSocketAddress, SocketConnection, SocketProtocol, SocketService, SocketType,
+};
+use log::info;
+
+async fn accept_connection(stream: SocketConnection) -> Result<()> {
+    let addr = stream
+        .socket()
+        .remote_address()
+        .expect("SocketConnection should have a remote address");
+
+    println!("Peer address: {}", addr);
+    let mut ws_stream = accept_async(stream)
+        .await
+        .expect("Error during the websocket handshake occurred");
+
+    while let Some(msg) = ws_stream.next().await {
+        let msg = msg?;
+        if msg.is_text() || msg.is_binary() {
+            ws_stream.send(msg).await?;
+        }
+    }
+
+    Ok(())
+}
+
+fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let _ = env_logger::try_init();
+    let addr = env::args()
+        .nth(1)
+        .unwrap_or_else(|| "127.0.0.1:8080".to_string());
+
+    let sockaddr: SocketAddr = addr.parse()?;
+    let inetaddr: InetSocketAddress = sockaddr.into();
+
+    let service = SocketService::new();
+    service.add_address(
+        &inetaddr,
+        SocketType::Stream,
+        SocketProtocol::Tcp,
+        glib::Object::NONE,
+    )?;
+    println!("Listening on: {}", inetaddr);
+
+    service.connect_incoming(|_service, connection, _| {
+        let stream = connection.clone();
+        glib::MainContext::default().spawn_local(async move {
+            accept_connection(stream).await;
+        });
+        false
+    });
+
+    let main_loop = glib::MainLoop::new(None, false);
+    main_loop.run();
+
+    Ok(())
+}

examples/gio-echo.rs

@@ -15,6 +15,8 @@ async fn run() -> Result<(), Box<dyn std::error::Error>> {
 
     println!("Received: {:?}", msg);
 
+    ws_stream.close(None).await.expect("error sending close");
+
     Ok(())
 }
 

src/gio.rs

@@ -9,6 +9,7 @@ use futures_io::{AsyncRead, AsyncWrite};
 
 use tungstenite::client::{uri_mode, IntoClientRequest};
 use tungstenite::handshake::client::Request;
+use tungstenite::handshake::server::{Callback, NoCallback};
 use tungstenite::stream::Mode;
 
 use crate::{client_async_with_config, domain, port, Response, WebSocketConfig, WebSocketStream};
@@ -61,6 +62,69 @@ where
     client_async_with_config(request, socket, config).await
 }
 
+/// Accepts a new WebSocket connection with the provided stream.
+///
+/// This function will internally call `server::accept` to create a
+/// handshake representation and returns a future representing the
+/// resolution of the WebSocket handshake. The returned future will resolve
+/// to either `WebSocketStream<S>` or `Error` depending if it's successful
+/// or not.
+///
+/// This is typically used after a socket has been accepted from a
+/// `TcpListener`. That socket is then passed to this function to perform
+/// the server half of the accepting a client's websocket connection.
+pub async fn accept_async<S>(stream: S) -> Result<WebSocketStream<IOStreamAsyncReadWrite<S>>, Error>
+where
+    S: IsA<gio::IOStream> + Unpin,
+{
+    accept_hdr_async(stream, NoCallback).await
+}
+
+/// The same as `accept_async()` but the one can specify a websocket configuration.
+/// Please refer to `accept_async()` for more details.
+pub async fn accept_async_with_config<S>(
+    stream: S,
+    config: Option<WebSocketConfig>,
+) -> Result<WebSocketStream<IOStreamAsyncReadWrite<S>>, Error>
+where
+    S: IsA<gio::IOStream> + Unpin,
+{
+    accept_hdr_async_with_config(stream, NoCallback, config).await
+}
+
+/// Accepts a new WebSocket connection with the provided stream.
+///
+/// This function does the same as `accept_async()` but accepts an extra callback
+/// for header processing. The callback receives headers of the incoming
+/// requests and is able to add extra headers to the reply.
+pub async fn accept_hdr_async<S, C>(
+    stream: S,
+    callback: C,
+) -> Result<WebSocketStream<IOStreamAsyncReadWrite<S>>, Error>
+where
+    S: IsA<gio::IOStream> + Unpin,
+    C: Callback + Unpin,
+{
+    accept_hdr_async_with_config(stream, callback, None).await
+}
+
+/// The same as `accept_hdr_async()` but the one can specify a websocket configuration.
+/// Please refer to `accept_hdr_async()` for more details.
+pub async fn accept_hdr_async_with_config<S, C>(
+    stream: S,
+    callback: C,
+    config: Option<WebSocketConfig>,
+) -> Result<WebSocketStream<IOStreamAsyncReadWrite<S>>, Error>
+where
+    S: IsA<gio::IOStream> + Unpin,
+    C: Callback + Unpin,
+{
+    let stream = IOStreamAsyncReadWrite::new(stream)
+        .map_err(|_| io::Error::new(io::ErrorKind::Other, "Unsupported gio::IOStream"))?;
+
+    crate::accept_hdr_async_with_config(stream, callback, config).await
+}
+
 /// Adapter for `gio::IOStream` to provide `AsyncRead` and `AsyncWrite`.
 #[derive(Debug)]
 pub struct IOStreamAsyncReadWrite<T: IsA<gio::IOStream>> {