export function createSocket (family: number): Socket {
  const socket = createUDP(family === 6 ? 'udp6' : 'udp4')
  const state = new SocketState()

  socket.once('close', () => {
    state.destroyed = true
    socket.removeAllListeners()
  })
  Object.assign(socket, { [kState]: state })
  return socket as Socket
}

export async function udpConnectionRequest(
  address: string,
  authExp: Expression,
  context: {},
  sourcePort: number = 0
): Promise<void> {
  const [host, portStr] = address.split(":", 2);
  const port = portStr ? parseInt(portStr) : 80;
  const now = Date.now();

  const client = dgram.createSocket({ type: "udp4", reuseAddr: true });
  // When a device is NAT'ed, the UDP Connection Request must originate from
  // the same address and port used by the STUN server, in order to traverse
  // the firewall. This does require that the Genieacs NBI and STUN server
  // are allowed to bind to the same address and port. The STUN server needs
  // to open its UDP port with the SO_REUSEADDR option, allowing the NBI to
  // also bind to the same port.
  if (sourcePort) client.bind({ port: sourcePort, exclusive: true });

  let username: string;
  let password: string;

  [username, password, authExp] = extractAuth(authExp, context, now, null);
  if (username == null) username = "";
  if (password == null) password = "";
  while (username != null && password != null) {
    const ts = Math.trunc(now / 1000);
    const id = Math.trunc(Math.random() * 4294967295);
    const cn = crypto.randomBytes(8).toString("hex");
    const sig = crypto
      .createHmac("sha1", password)
      .update(`${ts}${id}${username}${cn}`)
      .digest("hex");
    const uri = `http://${address}?ts=${ts}&id=${id}&un=${username}&cn=${cn}&sig=${sig}`;
    const message = Buffer.from(
      `GET ${uri} HTTP/1.1\r\nHost: ${address}\r\n\r\n`
    );

    for (let i = 0; i < 3; ++i) {
      await new Promise((resolve, reject) => {
        client.send(message, 0, message.length, port, host, (err: Error) => {
          if (err) reject(err);
          else resolve();
        });
      });
    }

    [username, password, authExp] = extractAuth(authExp, context, now, null);
  }
  client.close();
}

  static udpAssociate(client: net.Socket, dst: { port: number, addr: string }) {
    let udpType = 'udp' + (net.isIP(dst.addr) || 4);
    let serverUdp = dgram.createSocket(udpType);
    serverUdp.bind();
    serverUdp.unref();
    serverUdp.on('listening', async () => {
      let udpAddr = serverUdp.address();
      let reply = socks5Helper.createSocks5TcpReply(0x0, udpAddr.family === 'IPv4' ? ATYP.IPV4 : ATYP.IPV6, udpAddr.address, udpAddr.port);
      await client.writeAsync(reply);
    });
    
    let udpSet = new Set<dgram.Socket>();
    serverUdp.on('message', (msg: Buffer, rinfo: dgram.RemoteInfo) => {
      let socketId = `${rinfo.address}:${rinfo.port}`;
      let dst = socks5Helper.refineDestination(msg);
      
      let proxyUdp = dgram.createSocket(udpType);
      proxyUdp.unref();
      
      proxyUdp.on('message', (msg: Buffer) => {
        let header = socks5Helper.createSocks5UdpHeader(rinfo.address, rinfo.port);
        let data = Buffer.concat([header, msg]);
        serverUdp.send(data, 0, data.length, rinfo.port, rinfo.address);
      });
      
      proxyUdp.on('error', (err) => console.log(err.message));

      proxyUdp.send(msg, dst.headerSize, msg.length - dst.headerSize, dst.port, dst.addr);
      udpSet.add(proxyUdp);
    });
    
    function dispose() {
      client.dispose();
      
      serverUdp.removeAllListeners();
      serverUdp.close();
      
      udpSet.forEach(udp => {
        udp.close();
        udp.removeAllListeners();
      });
      
      udpSet.clear();
    }
    
    serverUdp.on('error', dispose);
    client.on('error', dispose);
    client.on('end', dispose);
  }

wss.on('connection', ws => {
  let tcpClient: net.Socket | undefined
  const udpClient = dgram.createSocket('udp4')
  udpClient.on('message', (msg: Buffer, rinfo: dgram.AddressInfo) => {
    ws.send(msg)
  })

  ws.on('message', data => {
    const protocol: types.Protocol = JSON.parse(data)
    if (protocol.kind === types.ProtocolKind.tcpConnect) {
      if (tcpClient) {
        tcpClient.destroy()
      }
      tcpClient = net.connect(protocol.port, protocol.host, () => {
        const responseProtocol: types.Protocol = {
          kind: types.ProtocolKind.tcpConnected
        }
        ws.send(JSON.stringify(responseProtocol))
      })
      tcpClient.on('close', hadError => {
        const responseProtocol: types.Protocol = {
          kind: types.ProtocolKind.tcpDisconnected
        }
        ws.send(JSON.stringify(responseProtocol))
      })
      tcpClient.on('error', error => {
        ws.send(`errored: ${error.stack}`)
      })
      tcpClient.on('timeout', () => {
        ws.send('timeout')
      })
      tcpClient.on('data', (tcpData: Buffer) => {
        ws.send(tcpData, { binary: true })
      })
    } else if (protocol.kind === types.ProtocolKind.tcpDisconnect) {
      if (tcpClient) {
        tcpClient.destroy()
      }
    } else if (protocol.kind === types.ProtocolKind.tcpSend) {
      if (tcpClient) {
        const message = protocol.isBinary ? new Buffer(protocol.message.split(',')) : protocol.message
        tcpClient.write(message)
      }
    } else if (protocol.kind === types.ProtocolKind.udpSend) {
      const message = protocol.isBinary ? new Buffer(protocol.message.split(',')) : protocol.message
      udpClient.send(message, protocol.port, protocol.address)
    }
  })
})

export function handleUDP(client: net.Socket, handshake: VpnHandshake, options: HandshakeOptions) {
  let communicationPending = false;
  let udpType = handshake.ipVer == IP_VER.V4 ? 'udp4' : 'udp6';
  let destAddress = handshake.destHost;
  let decipher: crypto.Decipher = null;
  
  let udpSocket = dgram.createSocket(udpType, async (msg: Buffer, rinfo: dgram.RemoteInfo) => {
    let iv = crypto.randomBytes(options.ivLength);
    let cipher = cryptoEx.createCipher(options.cipherAlgorithm, options.password, iv).cipher;
    let len = new Buffer(2);
    len.writeUInt16LE(msg.length, 0);
    let encrypted = cipher.update(Buffer.concat([len, msg]));
    await client.writeAsync(Buffer.concat([iv, encrypted]));
    communicationPending = true;
  });
  
  udpSocket.on('error', () => dispose());
  udpSocket.send(handshake.extra, 0, handshake.extra.length, handshake.destPort, destAddress);
  
  client.on('data', (d: Buffer) => {
    if (!decipher) decipher = cryptoEx.createDecipher(options.cipherAlgorithm, options.password, options.iv);
    let msg = decipher.update(d);
    udpSocket.send(msg, 0, msg.length, handshake.destPort, destAddress);
    communicationPending = true;
  });
  
  let cleanTimer = setInterval(() => {
    if (communicationPending) {
      communicationPending = false;
      return;
    }
    
    dispose();
  }, 30 * 1000);
  
  function dispose() {
    clearInterval(cleanTimer);
    client.dispose();
    udpSocket.close();
    udpSocket.unref();
    udpSocket.removeAllListeners();
  }
  
  client.on('error', () => dispose());
  client.on('end', () => dispose());
}

        it("should error observable if socket errors", (done) => {
            const server = dgram.createSocket("udp4");
            const observable = rxUdp.observableFromSocket<string>(
                b => b.toString("utf8"), server);

            observable.subscribeOnError(err => {
                expect(err).to.be.an("Error");
                expect(err.errno).to.equal("ENOTFOUND");
                expect(err.code).to.equal("ENOTFOUND");
                expect(err.syscall).to.equal("getaddrinfo");
                expect(err.hostname).to.equal("INEXISTENT_HOST");
                done();
            });

            // should result in error ENOTFOUND being emitted 
            server.bind(SERVER_PORT, "INEXISTENT_HOST");
        });

        (err, socket, info) => {
          if (err) {
            reject(new errors.RemoteUdpForwardingDisabled(`could not connect to local proxy`));
            return;
          }
          const packet = socks.createUDPFrame({host: '1.1.1.1', port: 53}, DNS_REQUEST);
          const udpSocket = dgram.createSocket('udp4');

          udpSocket.on('error', (e) => {
            reject(new errors.RemoteUdpForwardingDisabled('UDP socket failure'));
          });

          udpSocket.on('message', (msg, info) => {
            stopUdp();
            resolve(true);
          });

          // Retry sending the query every second.
          // TODO: logging here is a bit verbose
          const intervalId = setInterval(() => {
            try {
              udpSocket.send(packet, info.port, info.host, (err) => {
                if (err) {
                  console.error(`Failed to send data through UDP: ${err}`);
                }
              });
            } catch (e) {
              console.error(`Failed to send data through UDP ${e}`);
            }
          }, UDP_FORWARDING_TEST_RETRY_INTERVAL_MS);

          const stopUdp = () => {
            try {
              clearInterval(intervalId);
              udpSocket.close();
            } catch (e) {
              // Ignore; there may be multiple calls to this function.
            }
          };

          // Give up after the timeout elapses.
          setTimeout(() => {
            stopUdp();
            resolve(false);
          }, UDP_FORWARDING_TEST_TIMEOUT_MS);
        });

function test_statsd_client() {
  // can create client with defaults
  let client = new datadog.StatsD('localhost');
  let options: datadog.StatsDOptions = { global_tags: ['environment:definitely_typed']};
  const socket: dgram.Socket = dgram.createSocket('udp4');

  // can create client with all params
  client = new datadog.StatsD('localhost', 8125, socket, options);

  let key: string = 'key';
  let timeValue: number = 99;
  let sampleRate: number = 0.85;
  let incrementBy: number = 7;
  let decrementBy: number = 5;
  let gaugeValue: number = 199;
  let tags: string[] = ['tag1', 'tag2'];

  client.timing(key, timeValue);
  client.timing(key, timeValue, sampleRate);
  client.timing(key, timeValue, sampleRate, tags);

  client.increment(key);
  client.increment(key, sampleRate);
  client.increment(key, sampleRate, tags);

  client.incrementBy(key, incrementBy);
  client.incrementBy(key, incrementBy, tags);

  client.decrement(key);
  client.decrement(key, sampleRate);
  client.decrement(key, sampleRate, tags);

  client.decrementBy(key, decrementBy);
  client.decrementBy(key, decrementBy, tags);

  client.gauge(key, gaugeValue);
  client.gauge(key, gaugeValue, sampleRate);
  client.gauge(key, gaugeValue, sampleRate, tags);

  client.histogram(key, timeValue);
  client.histogram(key, timeValue, sampleRate);
  client.histogram(key, timeValue, sampleRate, tags);

  client.close();
}

    serverUdp.on('message', (msg: Buffer, rinfo: dgram.RemoteInfo) => {
      let socketId = `${rinfo.address}:${rinfo.port}`;
      let dst = socks5Helper.refineDestination(msg);
      
      let proxyUdp = dgram.createSocket(udpType);
      proxyUdp.unref();
      
      proxyUdp.on('message', (msg: Buffer) => {
        let header = socks5Helper.createSocks5UdpHeader(rinfo.address, rinfo.port);
        let data = Buffer.concat([header, msg]);
        serverUdp.send(data, 0, data.length, rinfo.port, rinfo.address);
      });
      
      proxyUdp.on('error', (err) => console.log(err.message));

      proxyUdp.send(msg, dst.headerSize, msg.length - dst.headerSize, dst.port, dst.addr);
      udpSet.add(proxyUdp);
    });

    return new Promise((resolve) => {
      this.socket = createSocket('udp4');

      this.socket.on('listening', () => {
        this.socket.setBroadcast(true);
        this.socket.setMulticastTTL(config.get<number>(MulticastServer.TTL));
        this.socket.addMembership(config.get<string>(MulticastServer.ADDRESS));

        resolve();
      });

      this.socket.on('message', (message: Buffer, remote: RemoteInfo) => {
        if(this.onMessageReceivedCallback) {
          this.onMessageReceivedCallback(message.toString(), remote);
        }
      });

      this.socket.bind(config.get<number>(MulticastServer.PORT));
    });