update

libxsync

@@ -1 +1 @@
-Subproject commit da1c2b91ecf79de858132d1eca04da7569961ac1
+Subproject commit 8433478652974055656b7f257047f654077bc5f8

mainwindow.cpp

@@ -123,14 +123,13 @@ MainWindow::~MainWindow() { delete ui; }
 void MainWindow::on_RefreshRegsButton_clicked() {  //
   ZLOGI(TAG, "on_refreshRegsButton_clicked");
 
-  // QtConcurrent::run([this]() {
-  //   while (true) {
-  //     ZLOGI(TAG, "on_refreshRegsButton_clicked......");
-  //     ZQThread::sleep(1);
-  //     auto ecode = Xsync::Ins().reg_write(0xABCDDCBA, 0x12345678);
-  //     ZLOGI(TAG, "reg_read ecode:%s", xs_error_code_2_str(ecode));
-  //   }
-  // });
+  QtConcurrent::run([this]() {
+    while (true) {
+      ZLOGI(TAG, "on_refreshRegsButton_clicked......");
+      auto ecode = Xsync::Ins().reg_write(0xABCDDCBA, 0x12345678);
+      ZLOGI(TAG, "reg_read ecode:%s", xs_error_code_2_str(ecode));
+    }
+  });
 
   // xs_error_code_t ecode = Xsync::Ins().reg_write(0xABCDDCBA, 0x12345678);
   // ZLOGI(TAG, "reg_write ecode:%s", xs_error_code_2_str(ecode));
@@ -146,10 +145,10 @@ void MainWindow::on_Connect2XsyncButton_clicked() {  //
 void MainWindow::mainWindowsRun() {  //
   XSyncUdpFactoryImpl::Ins()->initialize();
 
-  // auto xsudp = XSyncUdpFactoryImpl::Ins()->createXSUDP();
-  // XS_ASSERT(xsudp->initialize("0.0.0.0", 9999));
-  // xsudp->startReceive([this, xsudp](XsyncNetAdd &from, uint8_t *data, size_t length) {
+  auto xsudp = XSyncUdpFactoryImpl::Ins()->createXSUDP();
+  XS_ASSERT(xsudp->initialize("0.0.0.0", 9999));
+  xsudp->startReceive([this, xsudp](XsyncNetAdd &from, uint8_t *data, size_t length) {
     // ZLOGI(TAG, "receive from <%s:%d> (%d) :%s", from.ip.c_str(), from.port, data, length);
-  //   xsudp->sendto(from, "hello\n", 5, NULL);
-  // });
+    xsudp->sendto(from, "hello\n", 5, NULL);
+  });
 }

src/xsync_udp_factory_impl.cpp

@@ -11,6 +11,9 @@ using namespace iflytop;
 /*******************************************************************************
  *                                    XSUDP                                    *
  *******************************************************************************/
+
+#define USB_NO_BLOCK_UDP 0
+
 class XSUDP : public I_XSUDP {
   uint32_t m_ip;
   int      m_localport;
@@ -50,6 +53,13 @@ xs_error_code_t XSUDP::initialize(string ip, int localport) {
 
   int ret = bind(m_sock_fd, (struct sockaddr*)&localadd, sizeof(localadd));
   if (ret < 0) return kxs_ec_bind_fail;
+
+#if USB_NO_BLOCK_UDP
+  u_long mode = 1;
+  if (ioctlsocket(m_sock_fd, FIONBIO, &mode) != NO_ERROR) {
+    return kxs_ec_setsockopt_rx_timeout_fail;
+  }
+#endif
   return kxs_ec_success;
 }
 
@@ -66,26 +76,55 @@ xs_error_code_t XSUDP::sendto(const XsyncNetAdd& to, const char* data, int32_t l
   return kxs_ec_send_fail;
 }
 xs_error_code_t XSUDP::receive(char* data, int32_t& length, XsyncNetAdd& from, int overtimems) {
+#if USB_NO_BLOCK_UDP
+  struct sockaddr_in sockaddr      = {0};
+  int32_t            overtime_10ms = overtimems / 10;
+  int                ret           = 0;
+  overtime_10ms += 1;
+  for (size_t i = 0;; i++) {
+    int senderAddressLen = sizeof(sockaddr);
+    ret                  = ::recvfrom(m_sock_fd, (char*)data, length, 0, (struct sockaddr*)&sockaddr, &senderAddressLen);
+    length               = ret;
+
+    if (ret > 0) break;
+    if (i >= overtime_10ms) break;
+    Sleep(10);
+  }
+
+  if (ret <= 0) {
+    return kxs_ec_overtime;
+  }
+
+  uint32_t ip = ntohl(sockaddr.sin_addr.s_addr);
+  from.ip     = string(fmtip(ip));
+  from.port   = ntohs(sockaddr.sin_port);
+
+  return kxs_ec_success;
+#else
   struct sockaddr_in sockaddr = {0};
 
   timeval timeout;
-  // timeout.tv_sec  = overtimems / 1000;
-  timeout.tv_sec  = 1;
-  // timeout.tv_usec = overtimems % 1000 * 1000;
+  timeout.tv_sec  = overtimems / 1000;
   timeout.tv_usec = 0;
+
+  if (timeout.tv_sec == 0) {
+    timeout.tv_sec = 1;
+  }
+
   if (setsockopt(m_sock_fd, SOL_SOCKET, SO_RCVTIMEO, (const char*)&timeout, sizeof(timeout)) == -1) {
     return kxs_ec_setsockopt_rx_timeout_fail;
   }
-
   int senderAddressLen = sizeof(sockaddr);
   int ret              = ::recvfrom(m_sock_fd, (char*)data, length, 0, (struct sockaddr*)&sockaddr, &senderAddressLen);
   length               = ret;
+  data[length]         = 0;
+
   if (ret < 0) {
-    // if (errno == EWOULDBLOCK || errno == EAGAIN) {
+    if (errno == EWOULDBLOCK || errno == EAGAIN) {
       return kxs_ec_overtime;
-    // } else {
-    // return kxs_ec_receive_fail;
-    // }
+    } else {
+      return kxs_ec_receive_fail;
+    }
   }
 
   // inet_ntop(AF_INET, &(sockaddr.sin_addr), ip, INET_ADDRSTRLEN);
@@ -94,6 +133,7 @@ xs_error_code_t XSUDP::receive(char* data, int32_t& length, XsyncNetAdd& from, i
   from.port   = ntohs(sockaddr.sin_port);
 
   return kxs_ec_success;
+#endif
 }
 
 xs_error_code_t XSUDP::startReceive(onMessage_t onMessage) {
@@ -135,6 +175,7 @@ xs_error_code_t XSUDP::stopReceive() {
 }
 
 xs_error_code_t XSUDP::clearRxBuffer() {
+#if USB_NO_BLOCK_UDP
   char            buf[1024];
   int32_t         length = 1024;
   XsyncNetAdd     from;
@@ -143,6 +184,7 @@ xs_error_code_t XSUDP::clearRxBuffer() {
   while (ret == kxs_ec_success) {
     ret = receive(buf, length, from, 1);
   }
+#endif
 
   return kxs_ec_success;
 }