lzhr/v2fly
1
0
Fork 0
mirror of https://github.com/v2fly/v2ray-core.git synced 2025-01-18 07:17:32 -05:00
Code Issues Releases Wiki Activity

refactor

Browse Source
This commit is contained in:
Darien Raymond 2017-10-28 21:28:50 +02:00
parent 100391fdb7
commit ac5ab67862
1 changed files with 20 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

45
transport/internet/udp/hub.go
View File

@ -38,15 +38,16 @@ func NewPayloadQueue(option ListenOption) *PayloadQueue {
return queue return queue
} }
func (v *PayloadQueue) Enqueue(payload Payload) { // Enqueue adds the payload to the end of this queue.
size := len(v.queue) func (q *PayloadQueue) Enqueue(payload Payload) {
size := len(q.queue)
idx := 0 idx := 0
if size > 1 { if size > 1 {
idx = dice.Roll(size) idx = dice.Roll(size)
} }
for i := 0; i < size; i++ { for i := 0; i < size; i++ {
select { select {
case v.queue[idx%size] <- payload: case q.queue[idx%size] <- payload:
return return
default: default:
idx++ idx++
@ -54,14 +55,14 @@ func (v *PayloadQueue) Enqueue(payload Payload) {
} }
} }
func (v *PayloadQueue) Dequeue(queue <-chan Payload) { func (q *PayloadQueue) Dequeue(queue <-chan Payload) {
for payload := range queue { for payload := range queue {
v.callback(payload.payload, payload.source, payload.originalDest) q.callback(payload.payload, payload.source, payload.originalDest)
} }
} }
func (v *PayloadQueue) Close() { func (q *PayloadQueue) Close() {
for _, queue := range v.queue { for _, queue := range q.queue {
close(queue) close(queue)
} }
} }
@ -116,19 +117,19 @@ func ListenUDP(address net.Address, port net.Port, option ListenOption) (*Hub, e
return hub, nil return hub, nil
} }
func (v *Hub) Close() { func (h *Hub) Close() {
v.cancel() h.cancel()
v.conn.Close() h.conn.Close()
} }
func (v *Hub) WriteTo(payload []byte, dest net.Destination) (int, error) { func (h *Hub) WriteTo(payload []byte, dest net.Destination) (int, error) {
return v.conn.WriteToUDP(payload, &net.UDPAddr{ return h.conn.WriteToUDP(payload, &net.UDPAddr{
IP: dest.Address.IP(), IP: dest.Address.IP(),
Port: int(dest.Port), Port: int(dest.Port),
}) })
} }
func (v *Hub) start(ctx context.Context) { func (h *Hub) start(ctx context.Context) {
oobBytes := make([]byte, 256) oobBytes := make([]byte, 256)
L: L:
for { for {
@ -142,7 +143,7 @@ L:
var noob int var noob int
var addr *net.UDPAddr var addr *net.UDPAddr
err := buffer.AppendSupplier(func(b []byte) (int, error) { err := buffer.AppendSupplier(func(b []byte) (int, error) {
n, nb, _, a, e := ReadUDPMsg(v.conn, b, oobBytes) n, nb, _, a, e := ReadUDPMsg(h.conn, b, oobBytes)
noob = nb noob = nb
addr = a addr = a
return n, e return n, e
@ -158,20 +159,14 @@ L:
payload: buffer, payload: buffer,
} }
payload.source = net.UDPDestination(net.IPAddress(addr.IP), net.Port(addr.Port)) payload.source = net.UDPDestination(net.IPAddress(addr.IP), net.Port(addr.Port))
if v.option.ReceiveOriginalDest && noob > 0 { if h.option.ReceiveOriginalDest && noob > 0 {
payload.originalDest = RetrieveOriginalDest(oobBytes[:noob]) payload.originalDest = RetrieveOriginalDest(oobBytes[:noob])
} }
v.queue.Enqueue(payload) h.queue.Enqueue(payload)
} }
v.queue.Close() h.queue.Close()
} }
// Connection returns the net.Conn underneath this hub. func (h *Hub) Addr() net.Addr {
// Private: Visible for testing only return h.conn.LocalAddr()
func (v *Hub) Connection() net.Conn {
return v.conn
}
func (v *Hub) Addr() net.Addr {
return v.conn.LocalAddr()
} }