Merge pull request #9418 from ihnorton/udp_recvfrom

Add UDP recvfrom support (packet source addresses)

NEWS.md

@@ -145,6 +145,8 @@ Library improvements
 
   * A new `Val{T}` type allows one to dispatch on bits-type values ([#9452])
 
+  * Added `recvfrom` to get source address of UDP packets ([#9418])
+
 Deprecated or removed
 ---------------------
 

base/exports.jl

@@ -1177,6 +1177,7 @@ export
     redirect_stdin,
     redirect_stdout,
     recv,
+    recvfrom,
     reset,
     seek,
     seekend,

base/socket.jl

@@ -427,7 +427,7 @@ _bind(sock::UDPSocket, host::IPv6, port::UInt16, flags::UInt32 = uint32(0)) = cc
 
 function bind(sock::UDPSocket, host::IPv6, port::UInt16; ipv6only = false)
     @assert sock.status == StatusInit
-    err = _bind(sock,host,ipv6only ? UV_UDP_IPV6ONLY : 0)
+    err = _bind(sock,host,port, uint32(ipv6only ? UV_UDP_IPV6ONLY : 0))
     if err < 0
         if err != UV_EADDRINUSE && err != UV_EACCES
             error(UVError("bind",err))
@@ -470,22 +470,40 @@ end
 _recv_stop(sock::UDPSocket) = uv_error("recv_stop",ccall(:uv_udp_recv_stop,Cint,(Ptr{Void},),sock.handle))
 
 function recv(sock::UDPSocket)
+    addr, data = recvfrom(sock)
+    data
+end
+
+function recvfrom(sock::UDPSocket)
     # If the socket has not been bound, it will be bound implicitly to ::0 and a random port
     if sock.status != StatusInit && sock.status != StatusOpen
         error("Invalid socket state")
     end
     _recv_start(sock)
-    stream_wait(sock,sock.recvnotify)::Vector{UInt8}
+    stream_wait(sock,sock.recvnotify)::(Union(IPv4, IPv6), Vector{UInt8})
 end
 
+
 function _uv_hook_recv(sock::UDPSocket, nread::Int, buf_addr::Ptr{Void}, buf_size::UInt, addr::Ptr{Void}, flags::Int32)
+    # C signature documented as (*uv_udp_recv_cb)(...)
     if flags & UV_UDP_PARTIAL > 0
         # TODO: Decide what to do in this case. For now throw an error
         c_free(buf_addr)
         notify_error(sock.recvnotify,"Partial message received")
     end
+
+    # need to check the address type in order to convert to a Julia IPAddr
+    addrout = if (addr == C_NULL)
+                  IPv4(0)
+              elseif ccall(:jl_sockaddr_in_is_ip4, Cint, (Ptr{Void},), addr) == 1
+                  IPv4(ntoh(ccall(:jl_sockaddr_host4, Uint32, (Ptr{Void},), addr)))
+              else
+                  tmp = [uint128(0)]
+                  ccall(:jl_sockaddr_host6, Uint32, (Ptr{Void}, Ptr{Uint8}), addr, pointer(tmp))
+                  IPv6(ntoh(tmp[1]))
+              end
     buf = pointer_to_array(convert(Ptr{UInt8},buf_addr),int(buf_size),true)
-    notify(sock.recvnotify,buf[1:nread])
+    notify(sock.recvnotify,(addrout,buf[1:nread]))
 end
 
 function _send(sock::UDPSocket,ipaddr::IPv4,port::UInt16,buf)

doc/stdlib/io-network.rst

@@ -765,6 +765,10 @@ Network I/O
 
    Read a UDP packet from the specified socket, and return the bytes received. This call blocks.
 
+.. function:: recvfrom(socket::UDPSocket) -> (address, data)
+
+    Read a UDP packet from the specified socket, returning a tuple of (address, data), where address will be either IPv4 or IPv6 as appropriate.
+
 .. function:: setopt(sock::UDPSocket; multicast_loop = nothing, multicast_ttl=nothing, enable_broadcast=nothing, ttl=nothing)
 
    Set UDP socket options. ``multicast_loop``: loopback for multicast packets (default: true). ``multicast_ttl``: TTL for multicast packets. ``enable_broadcast``: flag must be set to true if socket will be used for broadcast messages, or else the UDP system will return an access error (default: false). ``ttl``: Time-to-live of packets sent on the socket.

test/socket.jl

@@ -73,25 +73,49 @@ close(server)
 
 @test_throws Base.UVError connect(".invalid",80)
 
-a = UDPSocket()
-b = UDPSocket()
-bind(a,ip"127.0.0.1",port)
-bind(b,ip"127.0.0.1",port+1)
+begin
+    a = UDPSocket()
+    b = UDPSocket()
+    bind(a,ip"127.0.0.1",port)
+    bind(b,ip"127.0.0.1",port+1)
 
-c = Condition()
-@async begin
-    @test bytestring(recv(a)) == "Hello World"
-    # Issue 6505
+    c = Condition()
     @async begin
         @test bytestring(recv(a)) == "Hello World"
-        notify(c)
+    # Issue 6505
+        @async begin
+            @test bytestring(recv(a)) == "Hello World"
+            notify(c)
+        end
+        send(b,ip"127.0.0.1",port,"Hello World")
     end
     send(b,ip"127.0.0.1",port,"Hello World")
-end
-send(b,ip"127.0.0.1",port,"Hello World")
-wait(c)
+    wait(c)
 
-@test_throws MethodError bind(UDPSocket(),port)
+    @async begin
+        @test begin
+            (addr,data) = recvfrom(a)
+            addr == ip"127.0.0.1" && bytestring(data) == "Hello World"
+        end
+    end
+    send(b, ip"127.0.0.1",port,"Hello World")
 
-close(a)
-close(b)
+    @test_throws MethodError bind(UDPSocket(),port)
+
+    close(a)
+    close(b)
+end
+begin
+    a = UDPSocket()
+    b = UDPSocket()
+    bind(a, ip"::1", uint16(port))
+    bind(b, ip"::1", uint16(port+1))
+
+    @async begin
+        @test begin
+            (addr, data) = recvfrom(a)
+            addr == ip"::1" && bytestring(data) == "Hello World"
+        end
+    end
+    send(b, ip"::1", port, "Hello World")
+end