Refactor WinRT socket to Win32 socket for Hotspot Server Socket for lower latency

PiperOrigin-RevId: 516905639

internal/platform/feature_flags.h

@@ -53,6 +53,11 @@ class FeatureFlags {
     // Controls to enable or disable to track the status of Bluetooth classic
     // conncetion.
     bool enable_bluetooth_connection_status_track = true;
+    // If Hotspot or WFD GO use WinRT API to create the server socket, the
+    // StreamSocketListener::BindEndpointAsync() will takes about 11s, so we
+    // prefer to use legacy WinSock API to create the server socket and do the
+    // binding, which only takes less than 0.1s
+    bool use_winsock = true;
   };
 
   static const FeatureFlags& GetInstance() {

internal/platform/implementation/windows/wifi_hotspot.h

@@ -16,11 +16,12 @@
 #define PLATFORM_IMPL_WINDOWS_WIFI_HOTSPOT_H_
 
 // Windows headers
+#include <winsock2.h>
 #include <windows.h>
 #include <wlanapi.h>
 
 // Standard C/C++ headers
-
+#include <deque>
 #include <memory>
 #include <optional>
 #include <string>
@@ -29,6 +30,7 @@
 #include "internal/platform/cancellation_flag_listener.h"
 #include "internal/platform/implementation/wifi_hotspot.h"
 #include "internal/platform/implementation/windows/scheduled_executor.h"
+#include "internal/platform/implementation/windows/submittable_executor.h"
 
 // WinRT headers
 #include "absl/types/optional.h"
@@ -91,12 +93,13 @@ using ::winrt::Windows::Networking::Sockets::
 
 // WifiHotspotSocket wraps the socket functions to read and write stream.
 // In WiFi HOTSPOT, A WifiHotspotSocket will be passed to
-// StartAcceptingConnections's call back when StreamSocketListener got connect.
-// When call API to connect to remote WiFi Hotspot service, also will return a
-// WifiHotspotSocket to caller.
+// StartAcceptingConnections's callback when Winsock Server Socket(or
+// StreamSocketListener) receives a new connection. When call API to connect to
+// remote WiFi Hotspot service, also will return a WifiHotspotSocket to caller.
 class WifiHotspotSocket : public api::WifiHotspotSocket {
  public:
   explicit WifiHotspotSocket(StreamSocket socket);
+  explicit WifiHotspotSocket(SOCKET socket);
   WifiHotspotSocket(const WifiHotspotSocket&) = default;
   WifiHotspotSocket(WifiHotspotSocket&&) = default;
   ~WifiHotspotSocket() override;
@@ -121,10 +124,12 @@ class WifiHotspotSocket : public api::WifiHotspotSocket {
   Exception Close() override;
 
  private:
+  enum class SocketType {kWinRTSocket = 0, kWin32Socket};
   // A simple wrapper to handle input stream of socket
   class SocketInputStream : public InputStream {
    public:
-    SocketInputStream(IInputStream input_stream);
+    explicit SocketInputStream(IInputStream input_stream);
+    explicit SocketInputStream(SOCKET socket);
     ~SocketInputStream() override = default;
 
     ExceptionOr<ByteArray> Read(std::int64_t size) override;
@@ -133,12 +138,15 @@ class WifiHotspotSocket : public api::WifiHotspotSocket {
 
    private:
     IInputStream input_stream_{nullptr};
+    SOCKET socket_ = INVALID_SOCKET;
+    SocketType socket_type_ = SocketType::kWinRTSocket;
   };
 
   // A simple wrapper to handle output stream of socket
   class SocketOutputStream : public OutputStream {
    public:
-    SocketOutputStream(IOutputStream output_stream);
+    explicit SocketOutputStream(IOutputStream output_stream);
+    explicit SocketOutputStream(SOCKET socket);
     ~SocketOutputStream() override = default;
 
     Exception Write(const ByteArray& data) override;
@@ -147,9 +155,12 @@ class WifiHotspotSocket : public api::WifiHotspotSocket {
 
    private:
     IOutputStream output_stream_{nullptr};
+    SOCKET socket_ = INVALID_SOCKET;
+    SocketType socket_type_ = SocketType::kWinRTSocket;
   };
 
   // Internal properties
+  SOCKET stream_soket_winsock_ = INVALID_SOCKET;
   StreamSocket stream_soket_{nullptr};
   SocketInputStream input_stream_{nullptr};
   SocketOutputStream output_stream_{nullptr};
@@ -198,6 +209,8 @@ class WifiHotspotServerSocket : public api::WifiHotspotServerSocket {
   fire_and_forget Listener_ConnectionReceived(
       StreamSocketListener listener,
       StreamSocketListenerConnectionReceivedEventArgs const& args);
+  bool SetupServerSocketWinRT();
+  bool SetupServerSocketWinSock();
 
   // Retrieves IP addresses from local machine
   std::vector<std::string> GetIpAddresses() const;
@@ -205,11 +218,15 @@ class WifiHotspotServerSocket : public api::WifiHotspotServerSocket {
 
   mutable absl::Mutex mutex_;
   absl::CondVar cond_;
+  SubmittableExecutor submittable_executor_;
 
   std::deque<StreamSocket> pending_sockets_ ABSL_GUARDED_BY(mutex_);
   StreamSocketListener stream_socket_listener_{nullptr};
   winrt::event_token listener_event_token_{};
 
+  std::deque<SOCKET> pending_client_sockets_ ABSL_GUARDED_BY(mutex_);
+  SOCKET listen_socket_ = INVALID_SOCKET;
+  SOCKET client_socket_ = INVALID_SOCKET;
   // Close notifier
   absl::AnyInvocable<void()> close_notifier_ = nullptr;
 

internal/platform/implementation/windows/wifi_hotspot_medium.cc

@@ -14,6 +14,7 @@
 
 #include <exception>
 #include <memory>
+#include <string>
 #include <utility>
 #include <vector>
 
@@ -197,8 +198,7 @@ WifiHotspotMedium::ListenForService(int port) {
 
   server_socket->SetCloseNotifier([this]() {
     absl::MutexLock lock(&mutex_);
-    NEARBY_LOGS(INFO) << "server socket was closed on port "
-                      << server_socket_ptr_->GetPort();
+    NEARBY_LOGS(INFO) << "Server socket was closed.";
     medium_status_ &= (~kMediumStatusAccepting);
     server_socket_ptr_ = nullptr;
   });

internal/platform/implementation/windows/wifi_hotspot_server_socket.cc

@@ -23,6 +23,7 @@
 #include "absl/strings/match.h"
 
 // Nearby connections headers
+#include "internal/platform/feature_flags.h"
 #include "internal/platform/implementation/windows/generated/winrt/Windows.Networking.Sockets.h"
 #include "internal/platform/implementation/windows/utils.h"
 #include "internal/platform/implementation/windows/wifi_hotspot.h"
@@ -35,7 +36,6 @@ using ::winrt::Windows::Networking::Sockets::SocketQualityOfService;
 
 constexpr int kMaxRetries = 3;
 constexpr int kRetryIntervalMilliSeconds = 300;
-
 }  // namespace
 
 WifiHotspotServerSocket::WifiHotspotServerSocket(int port) : port_(port) {}
@@ -43,8 +43,14 @@ WifiHotspotServerSocket::WifiHotspotServerSocket(int port) : port_(port) {}
 WifiHotspotServerSocket::~WifiHotspotServerSocket() { Close(); }
 
 std::string WifiHotspotServerSocket::GetIPAddress() const {
-  if (stream_socket_listener_ == nullptr) {
-    return {};
+  if (FeatureFlags::GetInstance().GetFlags().use_winsock) {
+    if (listen_socket_ == INVALID_SOCKET) {
+      return {};
+    }
+  } else {
+    if (stream_socket_listener_ == nullptr) {
+      return {};
+    }
   }
 
   if (ip_addresses_.empty()) {
@@ -58,17 +64,37 @@ std::string WifiHotspotServerSocket::GetIPAddress() const {
 }
 
 int WifiHotspotServerSocket::GetPort() const {
-  if (stream_socket_listener_ == nullptr) {
-    return 0;
+  if (FeatureFlags::GetInstance().GetFlags().use_winsock) {
+    if (listen_socket_ == INVALID_SOCKET) {
+      NEARBY_LOGS(WARNING) << __func__ << ": listen_socket_ is invalid.";
+      return 0;
+    }
+    return port_;
+  } else {
+    if (stream_socket_listener_ == nullptr) {
+      return 0;
+    }
+    return std::stoi(stream_socket_listener_.Information().LocalPort().c_str());
   }
-
-  return std::stoi(stream_socket_listener_.Information().LocalPort().c_str());
 }
 
 std::unique_ptr<api::WifiHotspotSocket> WifiHotspotServerSocket::Accept() {
   absl::MutexLock lock(&mutex_);
   NEARBY_LOGS(INFO) << __func__ << ": Accept is called.";
 
+  if (FeatureFlags::GetInstance().GetFlags().use_winsock) {
+    while (!closed_ && pending_client_sockets_.empty()) {
+      cond_.Wait(&mutex_);
+    }
+    if (closed_) return {};
+
+    SOCKET wifi_hotspot_socket = pending_client_sockets_.front();
+    pending_client_sockets_.pop_front();
+    NEARBY_LOGS(INFO) << __func__ << ": Accepted a remote connection.";
+    return std::make_unique<WifiHotspotSocket>(wifi_hotspot_socket);
+  }
+
+  // Code when using WinRT API
   while (!closed_ && pending_sockets_.empty()) {
     cond_.Wait(&mutex_);
   }
@@ -76,7 +102,6 @@ std::unique_ptr<api::WifiHotspotSocket> WifiHotspotServerSocket::Accept() {
 
   StreamSocket wifi_hotspot_socket = pending_sockets_.front();
   pending_sockets_.pop_front();
-
   NEARBY_LOGS(INFO) << __func__ << ": Accepted a remote connection.";
   return std::make_unique<WifiHotspotSocket>(wifi_hotspot_socket);
 }
@@ -91,20 +116,38 @@ Exception WifiHotspotServerSocket::Close() {
     absl::MutexLock lock(&mutex_);
     NEARBY_LOGS(INFO) << __func__ << ": Close is called.";
 
+    if (FeatureFlags::GetInstance().GetFlags().use_winsock) {
+      if (listen_socket_ != INVALID_SOCKET) {
+        NEARBY_LOGS(INFO) << ": Close listen_socket_: " << listen_socket_;
+        closesocket(listen_socket_);
+        closesocket(client_socket_);
+        listen_socket_ = INVALID_SOCKET;
+        client_socket_ = INVALID_SOCKET;
+        for (const auto &pending_socket : pending_client_sockets_) {
+          if (pending_socket != INVALID_SOCKET) closesocket(pending_socket);
+        }
+        WSACleanup();
+
+        pending_client_sockets_ = {};
+      }
+      submittable_executor_.Shutdown();
+    } else {
+      if (stream_socket_listener_ != nullptr) {
+        stream_socket_listener_.ConnectionReceived(listener_event_token_);
+        stream_socket_listener_.Close();
+        stream_socket_listener_ = nullptr;
+
+        for (const auto &pending_socket : pending_sockets_) {
+          pending_socket.Close();
+        }
+
+        pending_sockets_ = {};
+      }
+    }
+
     if (closed_) {
       return {Exception::kSuccess};
     }
-    if (stream_socket_listener_ != nullptr) {
-      stream_socket_listener_.ConnectionReceived(listener_event_token_);
-      stream_socket_listener_.Close();
-      stream_socket_listener_ = nullptr;
-
-      for (const auto &pending_socket : pending_sockets_) {
-        pending_socket.Close();
-      }
-
-      pending_sockets_ = {};
-    }
 
     closed_ = true;
     cond_.SignalAll();
@@ -133,25 +176,22 @@ Exception WifiHotspotServerSocket::Close() {
   }
 }
 
-bool WifiHotspotServerSocket::listen() {
-  // Get current IP addresses of the device.
-  for (int i = 0; i < kMaxRetries; i++) {
-    hotspot_ipaddr_ = GetHotspotIpAddresses();
-    if (hotspot_ipaddr_.empty()) {
-      NEARBY_LOGS(WARNING) << "Failed to find Hotspot's IP addr for the try: "
-                           << i + 1 << ". Wait " << kRetryIntervalMilliSeconds
-                           << "ms snd try again";
-      Sleep(kRetryIntervalMilliSeconds);
-    } else {
-      break;
-    }
-  }
-  if (hotspot_ipaddr_.empty()) {
-    NEARBY_LOGS(WARNING) << "Failed to start accepting connection without IP "
-                            "addresses configured on computer.";
-    return false;
+fire_and_forget WifiHotspotServerSocket::Listener_ConnectionReceived(
+    StreamSocketListener listener,
+    StreamSocketListenerConnectionReceivedEventArgs const &args) {
+  absl::MutexLock lock(&mutex_);
+  NEARBY_LOGS(INFO) << __func__ << ": Received connection.";
+
+  if (closed_) {
+    return fire_and_forget{};
   }
 
+  pending_sockets_.push_back(args.Socket());
+  cond_.SignalAll();
+  return fire_and_forget{};
+}
+
+bool WifiHotspotServerSocket::SetupServerSocketWinRT() {
   // Setup stream socket listener.
   stream_socket_listener_ = StreamSocketListener();
 
@@ -211,19 +251,159 @@ bool WifiHotspotServerSocket::listen() {
   return false;
 }
 
-fire_and_forget WifiHotspotServerSocket::Listener_ConnectionReceived(
-    StreamSocketListener listener,
-    StreamSocketListenerConnectionReceivedEventArgs const &args) {
-  absl::MutexLock lock(&mutex_);
-  NEARBY_LOGS(INFO) << __func__ << ": Received connection.";
+// Checks for SOCKET_ERROR, this error can come up when trying to bind, listen,
+// Getsockname, WSACreateEvent, WSAEventSelect etc.
+void SocketErrorNotice(SOCKET socket_to_close, const char *action) {
+  // const char *actionAttempted = action;
+  NEARBY_LOGS(WARNING) << "socket error. " << action
+                       << " failed with error: " << WSAGetLastError();
 
-  if (closed_) {
-    return fire_and_forget{};
+  closesocket(socket_to_close);
+  WSACleanup();
+}
+
+bool WifiHotspotServerSocket::SetupServerSocketWinSock() {
+  WSADATA wsa_data;
+  WSAEVENT socket_event;
+  int flag = 1;
+
+  int result = WSAStartup(MAKEWORD(2, 2), &wsa_data);
+  if (result != 0) {
+    NEARBY_LOGS(WARNING) << "WSAStartup failed with error:" << result;
+    return false;
   }
 
-  pending_sockets_.push_back(args.Socket());
-  cond_.SignalAll();
-  return fire_and_forget{};
+  listen_socket_ = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
+  if (listen_socket_ == INVALID_SOCKET) {
+    NEARBY_LOGS(WARNING) << "Failed to get socket";
+    WSACleanup();
+    return false;
+  }
+  struct sockaddr_in serv_addr;
+  serv_addr.sin_family = AF_INET;
+  serv_addr.sin_port = htons(port_);
+  serv_addr.sin_addr.s_addr = inet_addr(hotspot_ipaddr_.c_str());
+
+  unsigned long qos = 1;  // NOLINT
+  ioctlsocket(listen_socket_, SIO_SET_QOS, &qos);
+  setsockopt(listen_socket_, SOL_SOCKET, SO_KEEPALIVE, (const char *)&flag,
+             sizeof(flag));
+  if (bind(listen_socket_, (struct sockaddr *)&serv_addr, sizeof(serv_addr)) ==
+      SOCKET_ERROR) {
+    SocketErrorNotice(listen_socket_, "Bind");
+    return false;
+  }
+  NEARBY_LOGS(INFO) << "Bind socket successful";
+
+  int size = sizeof(serv_addr);
+  memset(&serv_addr, 0, size);
+  if (getsockname(listen_socket_, (struct sockaddr *)&serv_addr, &size) ==
+      SOCKET_ERROR) {
+    SocketErrorNotice(listen_socket_, "Getsockname");
+    return false;
+  }
+  port_ = ntohs(serv_addr.sin_port);
+  NEARBY_LOGS(INFO) << "Hotspot Server bound to port: " << port_;
+
+  socket_event = WSACreateEvent();
+  if (socket_event == nullptr) {
+    SocketErrorNotice(listen_socket_, "WSACreateEvent");
+    return false;
+  }
+  // Associate event types FD_ACCEPT and FD_CLOSE with the listen_socket_ and
+  // socket_event
+  if (WSAEventSelect(listen_socket_, socket_event, FD_ACCEPT | FD_CLOSE) ==
+      SOCKET_ERROR) {
+    SocketErrorNotice(listen_socket_, "WSAEventSelect");
+    return false;
+  }
+
+  if (::listen(listen_socket_, SOMAXCONN) == SOCKET_ERROR) {
+    SocketErrorNotice(listen_socket_, "Listen");
+    return false;
+  }
+  NEARBY_LOGS(INFO) << "Hotspot Server Socket " << listen_socket_
+                    << " started to listen with socket event: " << socket_event;
+
+  submittable_executor_.Execute([this, socket_event]() {
+    DWORD index;
+    WSANETWORKEVENTS network_events;
+    // Wait for network events on all sockets
+    index =
+        WSAWaitForMultipleEvents(1, &socket_event, FALSE, WSA_INFINITE, FALSE);
+
+    NEARBY_LOGS(INFO) << "Hotspot Server Socket " << listen_socket_
+                      << " received event: " << socket_event;
+    if (index == WSA_WAIT_TIMEOUT || index == WSA_WAIT_FAILED) {
+      NEARBY_LOGS(INFO) << "Hotspot Server Socket timout or failed ";
+      return false;
+    }
+
+    index = index - WSA_WAIT_EVENT_0;
+    // Iterate through all events and enumerate
+    if (WSAEnumNetworkEvents(listen_socket_, socket_event, &network_events) ==
+        SOCKET_ERROR) {
+      NEARBY_LOGS(INFO) << "Iterate through all events failed";
+      return false;
+    }
+    if (network_events.lNetworkEvents & FD_CLOSE) {
+      NEARBY_LOGS(INFO) << "Reveived FD_CLOSE event";
+      return false;
+    }
+    if (network_events.lNetworkEvents & FD_ACCEPT) {
+      client_socket_ = accept(listen_socket_, nullptr, nullptr);
+      NEARBY_LOGS(INFO) << "Reveived FD_ACCEPT event.";
+
+      if (client_socket_ == INVALID_SOCKET) {
+        return false;
+      }
+
+      if (WSAEventSelect(listen_socket_, socket_event, 0) == SOCKET_ERROR) {
+        NEARBY_LOGS(WARNING)
+            << "Remove association between listen_socket_ and event failed: "
+            << WSAGetLastError();
+      }
+
+      NEARBY_LOGS(INFO) << "Hotspot Server Client Socket created: "
+                        << client_socket_;
+      if (closed_) {
+        return false;
+      }
+      {
+        absl::MutexLock lock(&mutex_);
+        pending_client_sockets_.push_back(client_socket_);
+        cond_.SignalAll();
+      }
+    }
+    return true;
+  });
+  return true;
+}
+
+bool WifiHotspotServerSocket::listen() {
+  // Get current IP addresses of the device.
+  for (int i = 0; i < kMaxRetries; i++) {
+    hotspot_ipaddr_ = GetHotspotIpAddresses();
+    if (hotspot_ipaddr_.empty()) {
+      NEARBY_LOGS(WARNING) << "Failed to find Hotspot's IP addr for the try: "
+                           << i + 1 << ". Wait " << kRetryIntervalMilliSeconds
+                           << "ms snd try again";
+      Sleep(kRetryIntervalMilliSeconds);
+    } else {
+      break;
+    }
+  }
+  if (hotspot_ipaddr_.empty()) {
+    NEARBY_LOGS(WARNING) << "Failed to start accepting connection without IP "
+                            "addresses configured on computer.";
+    return false;
+  }
+
+  if (FeatureFlags::GetInstance().GetFlags().use_winsock) {
+    return SetupServerSocketWinSock();
+  } else {
+    return SetupServerSocketWinRT();
+  }
 }
 
 std::vector<std::string> WifiHotspotServerSocket::GetIpAddresses() const {

internal/platform/implementation/windows/wifi_hotspot_socket.cc

@@ -20,6 +20,11 @@
 
 namespace nearby {
 namespace windows {
+namespace {
+// Recorded the maximum bytes received in one read is 65586 in the test, but add
+// a little more for const kMaxByteRecieved for safty reason.
+constexpr int kMaxByteRecieved = 66000;
+}  // namespace
 
 WifiHotspotSocket::WifiHotspotSocket(StreamSocket socket) {
   stream_soket_ = socket;
@@ -27,9 +32,15 @@ WifiHotspotSocket::WifiHotspotSocket(StreamSocket socket) {
   output_stream_ = SocketOutputStream(socket.OutputStream());
 }
 
+WifiHotspotSocket::WifiHotspotSocket(SOCKET socket) {
+  stream_soket_winsock_ = socket;
+  input_stream_ = SocketInputStream(socket);
+  output_stream_ = SocketOutputStream(socket);
+}
+
 WifiHotspotSocket::~WifiHotspotSocket() {
   try {
-    if (stream_soket_ != nullptr) {
+    if (stream_soket_ != nullptr || stream_soket_winsock_ != INVALID_SOCKET) {
       Close();
     }
   } catch (std::exception exception) {
@@ -51,6 +62,10 @@ Exception WifiHotspotSocket::Close() {
     if (stream_soket_ != nullptr) {
       stream_soket_.Close();
     }
+    if (stream_soket_winsock_ != INVALID_SOCKET) {
+      closesocket(stream_soket_winsock_);
+      stream_soket_winsock_ = INVALID_SOCKET;
+    }
     return {Exception::kSuccess};
   } catch (std::exception exception) {
     NEARBY_LOGS(ERROR) << __func__ << ": Exception: " << exception.what();
@@ -68,23 +83,61 @@ Exception WifiHotspotSocket::Close() {
 WifiHotspotSocket::SocketInputStream::SocketInputStream(
     IInputStream input_stream) {
   input_stream_ = input_stream;
+  socket_type_ = SocketType::kWinRTSocket;
+}
+
+WifiHotspotSocket::SocketInputStream::SocketInputStream(SOCKET socket) {
+  socket_ = socket;
+  socket_type_ = SocketType::kWin32Socket;
 }
 
 ExceptionOr<ByteArray> WifiHotspotSocket::SocketInputStream::Read(
     std::int64_t size) {
   try {
-    Buffer buffer = Buffer(size);
+    if (socket_type_ == SocketType::kWinRTSocket) {
+      Buffer buffer = Buffer(size);
 
-    auto ibuffer =
-        input_stream_.ReadAsync(buffer, size, InputStreamOptions::None).get();
+      auto ibuffer =
+          input_stream_.ReadAsync(buffer, size, InputStreamOptions::None).get();
 
-    if (ibuffer.Length() != size) {
-      NEARBY_LOGS(WARNING) << "Only got part of data of needed.";
+      if (ibuffer.Length() != size) {
+        NEARBY_LOGS(WARNING) << "Only got part of data of needed.";
+      }
+
+      ByteArray data((char*)ibuffer.data(), ibuffer.Length());
+      return ExceptionOr<ByteArray>(data);
+    } else if (socket_type_ == SocketType::kWin32Socket) {
+      char recv_buf[kMaxByteRecieved];
+      int result;
+      struct fd_set readfds;
+
+      result = recv(socket_, recv_buf, size, 0);
+      if (result > 0) {
+        ByteArray data(recv_buf, result);
+        return ExceptionOr<ByteArray>(data);
+      } else if (result == 0) {
+        NEARBY_LOGS(INFO) << "Connection closed.";
+        return {Exception::kIo};
+      } else {
+        // When WSAEWOULDBLOCK happens, it means the packet for receive is not
+        // ready at the moment. The API select() will block till the packet is
+        // ready for recieving.
+        if (WSAEWOULDBLOCK == WSAGetLastError()) {
+          FD_ZERO(&readfds);
+          FD_SET(socket_, &readfds);
+          if (select(0, &readfds, nullptr, nullptr, nullptr) > 0) {
+            if (FD_ISSET(socket_, &readfds)) {
+              result = recv(socket_, recv_buf, size, 0);
+              if (result > 0) {
+                ByteArray data(recv_buf, result);
+                return ExceptionOr<ByteArray>(data);
+              }
+            }
+          }
+        }
+      }
     }
-
-    ByteArray data((char*)ibuffer.data(), ibuffer.Length());
-
-    return ExceptionOr(data);
+    return {Exception::kIo};
   } catch (std::exception exception) {
     NEARBY_LOGS(ERROR) << __func__ << ": Exception: " << exception.what();
     return {Exception::kIo};
@@ -100,11 +153,39 @@ ExceptionOr<ByteArray> WifiHotspotSocket::SocketInputStream::Read(
 
 ExceptionOr<size_t> WifiHotspotSocket::SocketInputStream::Skip(size_t offset) {
   try {
-    Buffer buffer = Buffer(offset);
+    if (socket_type_ == SocketType::kWinRTSocket) {
+      Buffer buffer = Buffer(offset);
 
-    auto ibuffer =
-        input_stream_.ReadAsync(buffer, offset, InputStreamOptions::None).get();
-    return ExceptionOr((size_t)ibuffer.Length());
+      auto ibuffer =
+          input_stream_.ReadAsync(buffer, offset, InputStreamOptions::None)
+              .get();
+      return ExceptionOr<size_t>((size_t)ibuffer.Length());
+    }
+    // When socket_type_ == SocketType::kWin32Socket
+    char recv_buf[kMaxByteRecieved];
+    int result;
+    struct fd_set readfds;
+
+    result = recv(socket_, recv_buf, offset, 0);
+    if (result > 0) {
+      return ExceptionOr<size_t>((size_t)result);
+    } else if (result == 0) {
+      NEARBY_LOGS(INFO) << "Connection closed.";
+    } else {
+      if (WSAEWOULDBLOCK == WSAGetLastError()) {
+        FD_ZERO(&readfds);
+        FD_SET(socket_, &readfds);
+        if (select(0, &readfds, nullptr, nullptr, nullptr) > 0) {
+          if (FD_ISSET(socket_, &readfds)) {
+            result = recv(socket_, recv_buf, offset, 0);
+            if (result > 0) {
+              return ExceptionOr<size_t>((size_t)result);
+            }
+          }
+        }
+      }
+    }
+    return {Exception::kIo};
   } catch (std::exception exception) {
     NEARBY_LOGS(ERROR) << __func__ << ": Exception: " << exception.what();
     return {Exception::kIo};
@@ -120,7 +201,12 @@ ExceptionOr<size_t> WifiHotspotSocket::SocketInputStream::Skip(size_t offset) {
 
 Exception WifiHotspotSocket::SocketInputStream::Close() {
   try {
-    input_stream_.Close();
+    if (socket_type_ == SocketType::kWinRTSocket) {
+      input_stream_.Close();
+    } else {
+      // When socket_type_ == SocketType::kWin32Socket
+      shutdown(socket_, SD_RECEIVE);
+    }
     return {Exception::kSuccess};
   } catch (std::exception exception) {
     NEARBY_LOGS(ERROR) << __func__ << ": Exception: " << exception.what();
@@ -139,16 +225,37 @@ Exception WifiHotspotSocket::SocketInputStream::Close() {
 WifiHotspotSocket::SocketOutputStream::SocketOutputStream(
     IOutputStream output_stream) {
   output_stream_ = output_stream;
+  socket_type_ = SocketType::kWinRTSocket;
+}
+
+WifiHotspotSocket::SocketOutputStream::SocketOutputStream(SOCKET socket) {
+  socket_ = socket;
+  socket_type_ = SocketType::kWin32Socket;
 }
 
 Exception WifiHotspotSocket::SocketOutputStream::Write(const ByteArray& data) {
   try {
-    Buffer buffer = Buffer(data.size());
-    std::memcpy(buffer.data(), data.data(), data.size());
-    buffer.Length(data.size());
+    if (socket_type_ == SocketType::kWinRTSocket) {
+      Buffer buffer = Buffer(data.size());
+      std::memcpy(buffer.data(), data.data(), data.size());
+      buffer.Length(data.size());
 
-    output_stream_.WriteAsync(buffer).get();
-    return {Exception::kSuccess};
+      output_stream_.WriteAsync(buffer).get();
+      return {Exception::kSuccess};
+    } else {
+      // When socket_type_ == SocketType::kWin32Socket
+      char sendbuf[kMaxByteRecieved];
+      int result;
+
+      std::memcpy(sendbuf, data.data(), data.size());
+      result = send(socket_, sendbuf, data.size(), 0);
+      if (result > 0) {
+        return {Exception::kSuccess};
+      } else {
+        NEARBY_LOGS(INFO) << "recv failed: " << WSAGetLastError();
+      }
+    }
+    return {Exception::kIo};
   } catch (std::exception exception) {
     NEARBY_LOGS(ERROR) << __func__ << ": Exception: " << exception.what();
     return {Exception::kIo};
@@ -164,7 +271,9 @@ Exception WifiHotspotSocket::SocketOutputStream::Write(const ByteArray& data) {
 
 Exception WifiHotspotSocket::SocketOutputStream::Flush() {
   try {
-    output_stream_.FlushAsync().get();
+    if (socket_type_ == SocketType::kWinRTSocket) {
+      output_stream_.FlushAsync().get();
+    }
     return {Exception::kSuccess};
   } catch (std::exception exception) {
     NEARBY_LOGS(ERROR) << __func__ << ": Exception: " << exception.what();
@@ -181,7 +290,11 @@ Exception WifiHotspotSocket::SocketOutputStream::Flush() {
 
 Exception WifiHotspotSocket::SocketOutputStream::Close() {
   try {
-    output_stream_.Close();
+    if (socket_type_ == SocketType::kWinRTSocket) {
+      output_stream_.Close();
+    } else if (socket_type_ == SocketType::kWin32Socket) {
+      shutdown(socket_, SD_SEND);
+    }
     return {Exception::kSuccess};
   } catch (std::exception exception) {
     NEARBY_LOGS(ERROR) << __func__ << ": Exception: " << exception.what();