ws
v1.4.0
Реализация RFC6455 WebSocket в Go.
wsutil , которые позволяют быстро начать работу, не копаясь во внутреннем устройстве протокола.ГоДок.
Существующие реализации WebSocket не позволяют пользователям повторно использовать буферы ввода-вывода между соединениями. Цель этой библиотеки — экспортировать эффективный низкоуровневый интерфейс для работы с протоколом, не ограничивая его использование только одним способом.
Кстати, если вы хотите получить инструменты более высокого уровня, вы можете использовать пакет wsutil .
Библиотека помечена как v1* поэтому ее API нельзя нарушать во время некоторых улучшений или рефакторинга.
Эта реализация RFC6455 проходит проверку Autobahn Test Suite и в настоящее время имеет покрытие около 78%.
Примеры приложений, использующих ws разрабатываются в отдельном репозитории ws-examples.
Пример более высокого уровня эхо-сервера WebSocket:
package main
import (
"net/http"
"github.com/gobwas/ws"
"github.com/gobwas/ws/wsutil"
)
func main () {
http . ListenAndServe ( ":8080" , http . HandlerFunc ( func ( w http. ResponseWriter , r * http. Request ) {
conn , _ , _ , err := ws . UpgradeHTTP ( r , w )
if err != nil {
// handle error
}
go func () {
defer conn . Close ()
for {
msg , op , err := wsutil . ReadClientData ( conn )
if err != nil {
// handle error
}
err = wsutil . WriteServerMessage ( conn , op , msg )
if err != nil {
// handle error
}
}
}()
}))
}Пример более низкого уровня, но все же высокого уровня:
import (
"net/http"
"io"
"github.com/gobwas/ws"
"github.com/gobwas/ws/wsutil"
)
func main () {
http . ListenAndServe ( ":8080" , http . HandlerFunc ( func ( w http. ResponseWriter , r * http. Request ) {
conn , _ , _ , err := ws . UpgradeHTTP ( r , w )
if err != nil {
// handle error
}
go func () {
defer conn . Close ()
var (
state = ws . StateServerSide
reader = wsutil . NewReader ( conn , state )
writer = wsutil . NewWriter ( conn , state , ws . OpText )
)
for {
header , err := reader . NextFrame ()
if err != nil {
// handle error
}
// Reset writer to write frame with right operation code.
writer . Reset ( conn , state , header . OpCode )
if _ , err = io . Copy ( writer , reader ); err != nil {
// handle error
}
if err = writer . Flush (); err != nil {
// handle error
}
}
}()
}))
}Мы можем применить тот же шаблон для чтения и записи структурированных ответов через кодировщик и декодер JSON:
...
var (
r = wsutil . NewReader ( conn , ws . StateServerSide )
w = wsutil . NewWriter ( conn , ws . StateServerSide , ws . OpText )
decoder = json . NewDecoder ( r )
encoder = json . NewEncoder ( w )
)
for {
hdr , err = r . NextFrame ()
if err != nil {
return err
}
if hdr . OpCode == ws . OpClose {
return io . EOF
}
var req Request
if err := decoder . Decode ( & req ); err != nil {
return err
}
var resp Response
if err := encoder . Encode ( & resp ); err != nil {
return err
}
if err = w . Flush (); err != nil {
return err
}
}
... Пример нижнего уровня без wsutil :
package main
import (
"net"
"io"
"github.com/gobwas/ws"
)
func main () {
ln , err := net . Listen ( "tcp" , "localhost:8080" )
if err != nil {
log . Fatal ( err )
}
for {
conn , err := ln . Accept ()
if err != nil {
// handle error
}
_ , err = ws . Upgrade ( conn )
if err != nil {
// handle error
}
go func () {
defer conn . Close ()
for {
header , err := ws . ReadHeader ( conn )
if err != nil {
// handle error
}
payload := make ([] byte , header . Length )
_ , err = io . ReadFull ( conn , payload )
if err != nil {
// handle error
}
if header . Masked {
ws . Cipher ( payload , header . Mask , 0 )
}
// Reset the Masked flag, server frames must not be masked as
// RFC6455 says.
header . Masked = false
if err := ws . WriteHeader ( conn , header ); err != nil {
// handle error
}
if _ , err := conn . Write ( payload ); err != nil {
// handle error
}
if header . OpCode == ws . OpClose {
return
}
}
}()
}
}Обновление с нулевым копированием помогает избежать ненужного выделения и копирования при обработке запроса на обновление HTTP.
Обработка всех заголовков, не относящихся к веб-сокетам, осуществляется с использованием зарегистрированных пользовательских обратных вызовов, аргументы которых действительны только до тех пор, пока обратный вызов не вернется.
Простой пример выглядит так:
package main
import (
"net"
"log"
"github.com/gobwas/ws"
)
func main () {
ln , err := net . Listen ( "tcp" , "localhost:8080" )
if err != nil {
log . Fatal ( err )
}
u := ws. Upgrader {
OnHeader : func ( key , value [] byte ) ( err error ) {
log . Printf ( "non-websocket header: %q=%q" , key , value )
return
},
}
for {
conn , err := ln . Accept ()
if err != nil {
// handle error
}
_ , err = u . Upgrade ( conn )
if err != nil {
// handle error
}
}
} Использование ws.Upgrader здесь дает возможность контролировать входящие соединения на уровне TCP и просто не принимать их по какой-то логике.
Обновление с нулевым копированием предназначено для служб с высокой нагрузкой, которым приходится контролировать множество ресурсов, таких как буферы соединений.
Пример из реальной жизни может быть таким:
package main
import (
"fmt"
"io"
"log"
"net"
"net/http"
"runtime"
"github.com/gobwas/httphead"
"github.com/gobwas/ws"
)
func main () {
ln , err := net . Listen ( "tcp" , "localhost:8080" )
if err != nil {
// handle error
}
// Prepare handshake header writer from http.Header mapping.
header := ws . HandshakeHeaderHTTP (http. Header {
"X-Go-Version" : [] string { runtime . Version ()},
})
u := ws. Upgrader {
OnHost : func ( host [] byte ) error {
if string ( host ) == "github.com" {
return nil
}
return ws . RejectConnectionError (
ws . RejectionStatus ( 403 ),
ws . RejectionHeader ( ws . HandshakeHeaderString (
"X-Want-Host: github.com r n " ,
)),
)
},
OnHeader : func ( key , value [] byte ) error {
if string ( key ) != "Cookie" {
return nil
}
ok := httphead . ScanCookie ( value , func ( key , value [] byte ) bool {
// Check session here or do some other stuff with cookies.
// Maybe copy some values for future use.
return true
})
if ok {
return nil
}
return ws . RejectConnectionError (
ws . RejectionReason ( "bad cookie" ),
ws . RejectionStatus ( 400 ),
)
},
OnBeforeUpgrade : func () (ws. HandshakeHeader , error ) {
return header , nil
},
}
for {
conn , err := ln . Accept ()
if err != nil {
log . Fatal ( err )
}
_ , err = u . Upgrade ( conn )
if err != nil {
log . Printf ( "upgrade error: %s" , err )
}
}
} Существует пакет ws/wsflate для поддержки расширения сжатия Permessage-Deflate.
Он предоставляет минималистичные оболочки ввода-вывода, которые можно использовать в сочетании с любой реализацией deflate (например, compress/flate стандартной библиотеки).
Он также совместим со средством чтения и записи wsutil , предоставляя тип wsflate.MessageState , который реализует интерфейсы wsutil.SendExtension и wsutil.RecvExtension .
package main
import (
"bytes"
"log"
"net"
"github.com/gobwas/ws"
"github.com/gobwas/ws/wsflate"
)
func main () {
ln , err := net . Listen ( "tcp" , "localhost:8080" )
if err != nil {
// handle error
}
e := wsflate. Extension {
// We are using default parameters here since we use
// wsflate.{Compress,Decompress}Frame helpers below in the code.
// This assumes that we use standard compress/flate package as flate
// implementation.
Parameters : wsflate . DefaultParameters ,
}
u := ws. Upgrader {
Negotiate : e . Negotiate ,
}
for {
conn , err := ln . Accept ()
if err != nil {
log . Fatal ( err )
}
// Reset extension after previous upgrades.
e . Reset ()
_ , err = u . Upgrade ( conn )
if err != nil {
log . Printf ( "upgrade error: %s" , err )
continue
}
if _ , ok := e . Accepted (); ! ok {
log . Printf ( "didn't negotiate compression for %s" , conn . RemoteAddr ())
conn . Close ()
continue
}
go func () {
defer conn . Close ()
for {
frame , err := ws . ReadFrame ( conn )
if err != nil {
// Handle error.
return
}
frame = ws . UnmaskFrameInPlace ( frame )
if wsflate . IsCompressed ( frame . Header ) {
// Note that even after successful negotiation of
// compression extension, both sides are able to send
// non-compressed messages.
frame , err = wsflate . DecompressFrame ( frame )
if err != nil {
// Handle error.
return
}
}
// Do something with frame...
ack := ws . NewTextFrame ([] byte ( "this is an acknowledgement" ))
// Compress response unconditionally.
ack , err = wsflate . CompressFrame ( ack )
if err != nil {
// Handle error.
return
}
if err = ws . WriteFrame ( conn , ack ); err != nil {
// Handle error.
return
}
}
}()
}
} Вы можете использовать сжатие с помощью пакета wsutil следующим образом:
// Upgrade somehow and negotiate compression to get the conn...
// Initialize flate reader. We are using nil as a source io.Reader because
// we will Reset() it in the message i/o loop below.
fr := wsflate . NewReader ( nil , func ( r io. Reader ) wsflate. Decompressor {
return flate . NewReader ( r )
})
// Initialize flate writer. We are using nil as a destination io.Writer
// because we will Reset() it in the message i/o loop below.
fw := wsflate . NewWriter ( nil , func ( w io. Writer ) wsflate. Compressor {
f , _ := flate . NewWriter ( w , 9 )
return f
})
// Declare compression message state variable.
//
// It has two goals:
// - Allow users to check whether received message is compressed or not.
// - Help wsutil.Reader and wsutil.Writer to set/unset appropriate
// WebSocket header bits while writing next frame to the wire (it
// implements wsutil.RecvExtension and wsutil.SendExtension).
var msg wsflate. MessageState
// Initialize WebSocket reader as previously.
// Please note the use of Reader.Extensions field as well as
// of ws.StateExtended flag.
rd := & wsutil. Reader {
Source : conn ,
State : ws . StateServerSide | ws . StateExtended ,
Extensions : []wsutil. RecvExtension {
& msg ,
},
}
// Initialize WebSocket writer with ws.StateExtended flag as well.
wr := wsutil . NewWriter ( conn , ws . StateServerSide | ws . StateExtended , 0 )
// Use the message state as wsutil.SendExtension.
wr . SetExtensions ( & msg )
for {
h , err := rd . NextFrame ()
if err != nil {
// handle error.
}
if h . OpCode . IsControl () {
// handle control frame.
}
if ! msg . IsCompressed () {
// handle uncompressed frame (skipped for the sake of example
// simplicity).
}
// Reset the writer to echo same op code.
wr . Reset ( h . OpCode )
// Reset both flate reader and writer to start the new round of i/o.
fr . Reset ( rd )
fw . Reset ( wr )
// Copy whole message from reader to writer decompressing it and
// compressing again.
if _ , err := io . Copy ( fw , fr ); err != nil {
// handle error.
}
// Flush any remaining buffers from flate writer to WebSocket writer.
if err := fw . Close (); err != nil {
// handle error.
}
// Flush the whole WebSocket message to the wire.
if err := wr . Flush (); err != nil {
// handle error.
}
}
