nats.io websocket / nats.io streaming websocket / nats.io websocket proxy

1-howto-nats-websocket-proxy.txt

on nats server run:
- start gnatsd server then:
-- go run natswsproxy.go -token test -bind :8888

on client run:
- go run client.go -to ws://ws.domain.tld:8888/mq?token=test -nats-user test -nats-pass test # user and pass from nats.server config

2-nats-ws-proxy.go

package main

import (
	"flag"
	"io"
	"log"
	"net"
	"net/http"
	"time"

	"github.com/gorilla/websocket"
)

var (
	flagBind    = flag.String("bind", ":8081", "bind to")
	flagBackend = flag.String("be", "127.0.0.1:4222", "nats server")
	flagToken   = flag.String("token", "test", "secret token for http connection")
	flagPath    = flag.String("http-path", "mq", "http path to websockets")
)
var upgrader = websocket.Upgrader{
	ReadBufferSize:  1024,
	WriteBufferSize: 1024,
}

func copyWorker(dst io.Writer, src io.Reader, doneCh chan bool) {
	io.Copy(dst, src)
	doneCh <- true
}

func main() {
	flag.Parse()
	upgrader.CheckOrigin = func(r *http.Request) bool { return true }
	http.HandleFunc("/"+*flagPath, func(w http.ResponseWriter, r *http.Request) {
		if r.URL.Query().Get("token") != *flagToken {
			http.NotFound(w, r)
			return
		}
		conn, err := upgrader.Upgrade(w, r, nil)
		if err != nil {
			log.Printf("%v\n", err)
			return
		}

		ncon, err := net.Dial("tcp", *flagBackend)
		if err != nil {
			log.Printf("%v", err)
			return
		}
		doneCh := make(chan bool)

		conn.UnderlyingConn().(*net.TCPConn).SetKeepAlivePeriod(1 * time.Second)

		go copyWorker(ncon, conn.UnderlyingConn(), doneCh)
		go copyWorker(conn.UnderlyingConn(), ncon, doneCh)

		<-doneCh

		ncon.Close()
		conn.Close()

		<-doneCh
	})

	http.ListenAndServe(*flagBind, nil)
}

3-client.go

package main

import (
	"flag"
	"log"
	"net"
	"net/url"
	"runtime"
	"time"

	nats "github.com/nats-io/nats.go"

	"github.com/gorilla/websocket"
)

var (
  flagURL  = flag.String("to", "wss://hostname.tld?token=test", "host to connect")
	flagUser = flag.String("nats-user", "", "nats user")
	flagPass = flag.String("nats-pass", "", "nats password")
)

type customDialer struct{}

func (cd *customDialer) Dial(network, address string) (net.Conn, error) {

	u, _ := url.Parse(*flagURL)
	c, _, err := websocket.DefaultDialer.Dial(u.String(), nil)
	if err != nil {
		return nil, err
	}
	return c.UnderlyingConn(), nil
}

func main() {
	flag.Parse()

	opts := []nats.Option{
		nats.SetCustomDialer(&customDialer{}),
		nats.ReconnectWait(1 * time.Second),
	}

	if *flagUser != "" {
		if *flagPass == "" {
			opts = append(opts, nats.Token(*flagUser))
		} else {
			opts = append(opts, nats.UserInfo(*flagUser, *flagPass))
		}
	}

	nc, err := nats.Connect(*flagURL, opts...)
	if err != nil {
		panic(err)
	}

	nc.Subscribe("test", func(msg *nats.Msg) {
		log.Printf("%s", msg.Data)
	})

	go func() {
		for {
			nc.Publish("test", []byte("test message"))
			time.Sleep(100 * time.Millisecond)
		}
	}()

	runtime.Goexit()

}