github/kubernetes
1
0
Fork 0
mirror of https://github.com/k3s-io/kubernetes.git synced 2025-07-24 12:15:52 +00:00
Code Issues Projects Releases Wiki Activity

Merge pull request #493 from lavalamp/scheduler

Browse Source 
Add websocket dep & watch api support
This commit is contained in:
Clayton Coleman 2014-07-18 22:18:03 -04:00
commit d1de579070
16 changed files with 2711 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
pkg/api/types.go
View File

@ -18,6 +18,7 @@ package api
import (
"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
"github.com/GoogleCloudPlatform/kubernetes/pkg/watch"
"github.com/fsouza/go-dockerclient"
)
@ -338,3 +339,12 @@ type ServerOpList struct {
JSONBase `yaml:",inline" json:",inline"`
Items []ServerOp `yaml:"items,omitempty" json:"items,omitempty"`
}
// WatchEvent objects are streamed from the api server in response to a watch request.
type WatchEvent struct {
// The type of the watch event; added, modified, or deleted.
Type watch.EventType
// An object which can be decoded via api.Decode
EmbeddedObject []byte
}

204
pkg/apiserver/apiserver.go
View File

@ -17,6 +17,7 @@ limitations under the License.
package apiserver
import (
"encoding/json"
"fmt"
"io/ioutil"
"net/http"
@ -25,12 +26,14 @@ import (
"strings"
"time"
"code.google.com/p/go.net/websocket"
"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
"github.com/GoogleCloudPlatform/kubernetes/pkg/healthz"
"github.com/GoogleCloudPlatform/kubernetes/pkg/httplog"
"github.com/GoogleCloudPlatform/kubernetes/pkg/labels"
"github.com/GoogleCloudPlatform/kubernetes/pkg/tools"
"github.com/GoogleCloudPlatform/kubernetes/pkg/util"
"github.com/GoogleCloudPlatform/kubernetes/pkg/watch"
"github.com/golang/glog"
)
@ -72,6 +75,13 @@ type RESTStorage interface {
Update(interface{}) (<-chan interface{}, error)
}
// ResourceWatcher should be implemented by all RESTStorage objects that
// want to offer the ability to watch for changes through the watch api.
type ResourceWatcher interface {
WatchAll() (watch.Interface, error)
WatchSingle(id string) (watch.Interface, error)
}
// WorkFunc is used to perform any time consuming work for an api call, after
// the input has been validated. Pass one of these to MakeAsync to create an
// appropriate return value for the Update, Delete, and Create methods.
@ -136,13 +146,24 @@ func New(storage map[string]RESTStorage, prefix string) *APIServer {
s.mux.HandleFunc("/index.html", s.handleIndex)
// Handle both operations and operations/* with the same handler
opPrefix := path.Join(s.prefix, "operations")
s.mux.HandleFunc(opPrefix, s.handleOperationRequest)
s.mux.HandleFunc(opPrefix+"/", s.handleOperationRequest)
s.mux.HandleFunc(s.operationPrefix(), s.handleOperationRequest)
s.mux.HandleFunc(s.operationPrefix()+"/", s.handleOperationRequest)
s.mux.HandleFunc(s.watchPrefix()+"/", s.handleWatch)
s.mux.HandleFunc("/", s.notFound)
return s
}
func (s *APIServer) operationPrefix() string {
return path.Join(s.prefix, "operations")
}
func (s *APIServer) watchPrefix() string {
return path.Join(s.prefix, "watch")
}
func (server *APIServer) handleIndex(w http.ResponseWriter, req *http.Request) {
w.WriteHeader(http.StatusOK)
// TODO: serve this out of a file?
@ -175,25 +196,25 @@ func (server *APIServer) ServeHTTP(w http.ResponseWriter, req *http.Request) {
// ServeREST handles requests to all our RESTStorage objects.
func (server *APIServer) ServeREST(w http.ResponseWriter, req *http.Request) {
if !strings.HasPrefix(req.URL.Path, server.prefix) {
server.notFound(req, w)
server.notFound(w, req)
return
}
requestParts := strings.Split(req.URL.Path[len(server.prefix):], "/")[1:]
if len(requestParts) < 1 {
server.notFound(req, w)
server.notFound(w, req)
return
}
storage := server.storage[requestParts[0]]
if storage == nil {
httplog.LogOf(w).Addf("'%v' has no storage object", requestParts[0])
server.notFound(req, w)
server.notFound(w, req)
return
}
server.handleREST(requestParts, req, w, storage)
}
func (server *APIServer) notFound(req *http.Request, w http.ResponseWriter) {
func (server *APIServer) notFound(w http.ResponseWriter, req *http.Request) {
w.WriteHeader(http.StatusNotFound)
fmt.Fprintf(w, "Not Found: %#v", req)
}
@ -290,7 +311,7 @@ func (server *APIServer) handleREST(parts []string, req *http.Request, w http.Re
case 2:
item, err := storage.Get(parts[1])
if IsNotFound(err) {
server.notFound(req, w)
server.notFound(w, req)
return
}
if err != nil {
@ -299,11 +320,11 @@ func (server *APIServer) handleREST(parts []string, req *http.Request, w http.Re
}
server.write(http.StatusOK, item, w)
default:
server.notFound(req, w)
server.notFound(w, req)
}
case "POST":
if len(parts) != 1 {
server.notFound(req, w)
server.notFound(w, req)
return
}
body, err := server.readBody(req)
@ -313,7 +334,7 @@ func (server *APIServer) handleREST(parts []string, req *http.Request, w http.Re
}
obj, err := storage.Extract(body)
if IsNotFound(err) {
server.notFound(req, w)
server.notFound(w, req)
return
}
if err != nil {
@ -322,7 +343,7 @@ func (server *APIServer) handleREST(parts []string, req *http.Request, w http.Re
}
out, err := storage.Create(obj)
if IsNotFound(err) {
server.notFound(req, w)
server.notFound(w, req)
return
}
if err != nil {
@ -332,12 +353,12 @@ func (server *APIServer) handleREST(parts []string, req *http.Request, w http.Re
server.finishReq(out, sync, timeout, w)
case "DELETE":
if len(parts) != 2 {
server.notFound(req, w)
server.notFound(w, req)
return
}
out, err := storage.Delete(parts[1])
if IsNotFound(err) {
server.notFound(req, w)
server.notFound(w, req)
return
}
if err != nil {
@ -347,7 +368,7 @@ func (server *APIServer) handleREST(parts []string, req *http.Request, w http.Re
server.finishReq(out, sync, timeout, w)
case "PUT":
if len(parts) != 2 {
server.notFound(req, w)
server.notFound(w, req)
return
}
body, err := server.readBody(req)
@ -357,7 +378,7 @@ func (server *APIServer) handleREST(parts []string, req *http.Request, w http.Re
}
obj, err := storage.Extract(body)
if IsNotFound(err) {
server.notFound(req, w)
server.notFound(w, req)
return
}
if err != nil {
@ -366,7 +387,7 @@ func (server *APIServer) handleREST(parts []string, req *http.Request, w http.Re
}
out, err := storage.Update(obj)
if IsNotFound(err) {
server.notFound(req, w)
server.notFound(w, req)
return
}
if err != nil {
@ -375,24 +396,24 @@ func (server *APIServer) handleREST(parts []string, req *http.Request, w http.Re
}
server.finishReq(out, sync, timeout, w)
default:
server.notFound(req, w)
server.notFound(w, req)
}
}
func (server *APIServer) handleOperationRequest(w http.ResponseWriter, req *http.Request) {
opPrefix := path.Join(server.prefix, "operations")
opPrefix := server.operationPrefix()
if !strings.HasPrefix(req.URL.Path, opPrefix) {
server.notFound(req, w)
server.notFound(w, req)
return
}
trimmed := strings.TrimLeft(req.URL.Path[len(opPrefix):], "/")
parts := strings.Split(trimmed, "/")
if len(parts) > 1 {
server.notFound(req, w)
server.notFound(w, req)
return
}
if req.Method != "GET" {
server.notFound(req, w)
server.notFound(w, req)
return
}
if len(parts) == 0 {
@ -404,7 +425,7 @@ func (server *APIServer) handleOperationRequest(w http.ResponseWriter, req *http
op := server.ops.Get(parts[0])
if op == nil {
server.notFound(req, w)
server.notFound(w, req)
return
}
@ -415,3 +436,140 @@ func (server *APIServer) handleOperationRequest(w http.ResponseWriter, req *http
server.write(http.StatusAccepted, obj, w)
}
}
func (server *APIServer) handleWatch(w http.ResponseWriter, req *http.Request) {
prefix := server.watchPrefix()
if !strings.HasPrefix(req.URL.Path, prefix) {
server.notFound(w, req)
return
}
parts := strings.Split(req.URL.Path[len(prefix):], "/")[1:]
if req.Method != "GET" || len(parts) < 1 {
server.notFound(w, req)
}
storage := server.storage[parts[0]]
if storage == nil {
server.notFound(w, req)
}
if watcher, ok := storage.(ResourceWatcher); ok {
var watching watch.Interface
var err error
if id := req.URL.Query().Get("id"); id != "" {
watching, err = watcher.WatchSingle(id)
} else {
watching, err = watcher.WatchAll()
}
if err != nil {
server.error(err, w)
return
}
// TODO: This is one watch per connection. We want to multiplex, so that
// multiple watches of the same thing don't create two watches downstream.
watchServer := &WatchServer{watching}
if req.Header.Get("Connection") == "Upgrade" && req.Header.Get("Upgrade") == "websocket" {
websocket.Handler(watchServer.HandleWS).ServeHTTP(httplog.Unlogged(w), req)
} else {
watchServer.ServeHTTP(w, req)
}
return
}
server.notFound(w, req)
}
// WatchServer serves a watch.Interface over a websocket or vanilla HTTP.
type WatchServer struct {
watching watch.Interface
}
// HandleWS implements a websocket handler.
func (w *WatchServer) HandleWS(ws *websocket.Conn) {
done := make(chan struct{})
go func() {
var unused interface{}
// Expect this to block until the connection is closed. Client should not
// send anything.
websocket.JSON.Receive(ws, &unused)
close(done)
}()
for {
select {
case <-done:
w.watching.Stop()
return
case event, ok := <-w.watching.ResultChan():
if !ok {
// End of results.
return
}
wireFormat, err := api.Encode(event.Object)
if err != nil {
glog.Errorf("error encoding %#v: %v", event.Object, err)
return
}
err = websocket.JSON.Send(ws, &api.WatchEvent{
Type: event.Type,
EmbeddedObject: wireFormat,
})
if err != nil {
// Client disconnect.
w.watching.Stop()
return
}
}
}
}
// ServeHTTP serves a series of JSON encoded events via straight HTTP with
// Transfer-Encoding: chunked.
func (self *WatchServer) ServeHTTP(w http.ResponseWriter, req *http.Request) {
loggedW := httplog.LogOf(w)
w = httplog.Unlogged(w)
cn, ok := w.(http.CloseNotifier)
if !ok {
loggedW.Addf("unable to get CloseNotifier")
http.NotFound(loggedW, req)
return
}
flusher, ok := w.(http.Flusher)
if !ok {
loggedW.Addf("unable to get Flusher")
http.NotFound(loggedW, req)
return
}
loggedW.Header().Set("Transfer-Encoding", "chunked")
loggedW.WriteHeader(http.StatusOK)
flusher.Flush()
encoder := json.NewEncoder(w)
for {
select {
case <-cn.CloseNotify():
self.watching.Stop()
return
case event, ok := <-self.watching.ResultChan():
if !ok {
// End of results.
return
}
wireFormat, err := api.Encode(event.Object)
if err != nil {
glog.Errorf("error encoding %#v: %v", event.Object, err)
return
}
err = encoder.Encode(&api.WatchEvent{
Type: event.Type,
EmbeddedObject: wireFormat,
})
if err != nil {
// Client disconnect.
self.watching.Stop()
return
}
flusher.Flush()
}
}
}

161
pkg/apiserver/apiserver_test.go
View File

@ -18,17 +18,21 @@ package apiserver
import (
"bytes"
"encoding/json"
"fmt"
"io/ioutil"
"net/http"
"net/http/httptest"
"net/url"
"reflect"
"sync"
"testing"
"time"
"code.google.com/p/go.net/websocket"
"github.com/GoogleCloudPlatform/kubernetes/pkg/api"
"github.com/GoogleCloudPlatform/kubernetes/pkg/labels"
"github.com/GoogleCloudPlatform/kubernetes/pkg/watch"
)
func init() {
@ -60,6 +64,12 @@ type SimpleRESTStorage struct {
updated Simple
created Simple
// Valid if WatchAll or WatchSingle is called
fakeWatch *watch.FakeWatcher
// Set if WatchSingle is called
requestedID string
// If non-nil, called inside the WorkFunc when answering update, delete, create.
// obj recieves the original input to the update, delete, or create call.
injectedFunction func(obj interface{}) (returnObj interface{}, err error)
@ -78,8 +88,8 @@ func (storage *SimpleRESTStorage) Get(id string) (interface{}, error) {
func (storage *SimpleRESTStorage) Delete(id string) (<-chan interface{}, error) {
storage.deleted = id
if storage.errors["delete"] != nil {
return nil, storage.errors["delete"]
if err := storage.errors["delete"]; err != nil {
return nil, err
}
return MakeAsync(func() (interface{}, error) {
if storage.injectedFunction != nil {
@ -97,8 +107,8 @@ func (storage *SimpleRESTStorage) Extract(body []byte) (interface{}, error) {
func (storage *SimpleRESTStorage) Create(obj interface{}) (<-chan interface{}, error) {
storage.created = obj.(Simple)
if storage.errors["create"] != nil {
return nil, storage.errors["create"]
if err := storage.errors["create"]; err != nil {
return nil, err
}
return MakeAsync(func() (interface{}, error) {
if storage.injectedFunction != nil {
@ -110,8 +120,8 @@ func (storage *SimpleRESTStorage) Create(obj interface{}) (<-chan interface{}, e
func (storage *SimpleRESTStorage) Update(obj interface{}) (<-chan interface{}, error) {
storage.updated = obj.(Simple)
if storage.errors["update"] != nil {
return nil, storage.errors["update"]
if err := storage.errors["update"]; err != nil {
return nil, err
}
return MakeAsync(func() (interface{}, error) {
if storage.injectedFunction != nil {
@ -121,6 +131,25 @@ func (storage *SimpleRESTStorage) Update(obj interface{}) (<-chan interface{}, e
}), nil
}
// Implement ResourceWatcher.
func (storage *SimpleRESTStorage) WatchAll() (watch.Interface, error) {
if err := storage.errors["watchAll"]; err != nil {
return nil, err
}
storage.fakeWatch = watch.NewFake()
return storage.fakeWatch, nil
}
// Implement ResourceWatcher.
func (storage *SimpleRESTStorage) WatchSingle(id string) (watch.Interface, error) {
storage.requestedID = id
if err := storage.errors["watchSingle"]; err != nil {
return nil, err
}
storage.fakeWatch = watch.NewFake()
return storage.fakeWatch, nil
}
func extractBody(response *http.Response, object interface{}) (string, error) {
defer response.Body.Close()
body, err := ioutil.ReadAll(response.Body)
@ -525,3 +554,123 @@ func TestOpGet(t *testing.T) {
t.Errorf("Unexpected response %#v", response)
}
}
var watchTestTable = []struct {
t watch.EventType
obj interface{}
}{
{watch.Added, &Simple{Name: "A Name"}},
{watch.Modified, &Simple{Name: "Another Name"}},
{watch.Deleted, &Simple{Name: "Another Name"}},
}
func TestWatchWebsocket(t *testing.T) {
simpleStorage := &SimpleRESTStorage{}
handler := New(map[string]RESTStorage{
"foo": simpleStorage,
}, "/prefix/version")
server := httptest.NewServer(handler)
dest, _ := url.Parse(server.URL)
dest.Scheme = "ws" // Required by websocket, though the server never sees it.
dest.Path = "/prefix/version/watch/foo"
dest.RawQuery = "id=myID"
ws, err := websocket.Dial(dest.String(), "", "http://localhost")
expectNoError(t, err)
if a, e := simpleStorage.requestedID, "myID"; a != e {
t.Fatalf("Expected %v, got %v", e, a)
}
try := func(action watch.EventType, object interface{}) {
// Send
simpleStorage.fakeWatch.Action(action, object)
// Test receive
var got api.WatchEvent
err := websocket.JSON.Receive(ws, &got)
if err != nil {
t.Fatalf("Unexpected error: %v", err)
}
if got.Type != action {
t.Errorf("Unexpected type: %v", got.Type)
}
apiObj, err := api.Decode(got.EmbeddedObject)
if err != nil {
t.Fatalf("Unexpected error: %v", err)
}
if !reflect.DeepEqual(object, apiObj) {
t.Errorf("Expected %v, got %v", object, apiObj)
}
}
for _, item := range watchTestTable {
try(item.t, item.obj)
}
simpleStorage.fakeWatch.Stop()
var got api.WatchEvent
err = websocket.JSON.Receive(ws, &got)
if err == nil {
t.Errorf("Unexpected non-error")
}
}
func TestWatchHTTP(t *testing.T) {
simpleStorage := &SimpleRESTStorage{}
handler := New(map[string]RESTStorage{
"foo": simpleStorage,
}, "/prefix/version")
server := httptest.NewServer(handler)
client := http.Client{}
dest, _ := url.Parse(server.URL)
dest.Path = "/prefix/version/watch/foo"
dest.RawQuery = "id=myID"
request, err := http.NewRequest("GET", dest.String(), nil)
expectNoError(t, err)
response, err := client.Do(request)
expectNoError(t, err)
if response.StatusCode != http.StatusOK {
t.Errorf("Unexpected response %#v", response)
}
if a, e := simpleStorage.requestedID, "myID"; a != e {
t.Fatalf("Expected %v, got %v", e, a)
}
decoder := json.NewDecoder(response.Body)
try := func(action watch.EventType, object interface{}) {
// Send
simpleStorage.fakeWatch.Action(action, object)
// Test receive
var got api.WatchEvent
err := decoder.Decode(&got)
if err != nil {
t.Fatalf("Unexpected error: %v", err)
}
if got.Type != action {
t.Errorf("Unexpected type: %v", got.Type)
}
apiObj, err := api.Decode(got.EmbeddedObject)
if err != nil {
t.Fatalf("Unexpected error: %v", err)
}
if !reflect.DeepEqual(object, apiObj) {
t.Errorf("Expected %v, got %v", object, apiObj)
}
}
for _, item := range watchTestTable {
try(item.t, item.obj)
}
simpleStorage.fakeWatch.Stop()
var got api.WatchEvent
err = decoder.Decode(&got)
if err == nil {
t.Errorf("Unexpected non-error")
}
}

8
pkg/httplog/log.go
View File

@ -96,6 +96,14 @@ func LogOf(w http.ResponseWriter) *respLogger {
return nil
}
// Unlogged returns the original ResponseWriter, or w if it is not our inserted logger.
func Unlogged(w http.ResponseWriter) http.ResponseWriter {
if rl, ok := w.(*respLogger); ok {
return rl.w
}
return w
}
// Sets the stacktrace logging predicate, which decides when to log a stacktrace.
// There's a default, so you don't need to call this unless you don't like the default.
func (rl *respLogger) StacktraceWhen(pred StacktracePred) *respLogger {

19
pkg/watch/doc.go Normal file
View File

@ -0,0 +1,19 @@
/*
Copyright 2014 Google Inc. All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
// Package watch contains a generic watchable interface, and a fake for
// testing code that uses the watch interface.
package watch

96
pkg/watch/watch.go Normal file
View File

@ -0,0 +1,96 @@
/*
Copyright 2014 Google Inc. All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package watch
import (
"sync"
)
// Interface can be implemented by anything that knows how to watch and report changes.
type Interface interface {
// Stops watching. Will close the channel returned by ResultChan(). Releases
// any resources used by the watch.
Stop()
// Returns a chan which will receive all the events. If an error occurs
// or Stop() is called, this channel will be closed, in which case the
// watch should be completely cleaned up.
ResultChan() <-chan *Event
}
// EventType defines the possible types of events.
type EventType string
const (
Added EventType = "ADDED"
Modified EventType = "MODIFIED"
Deleted EventType = "DELETED"
)
// Event represents a single event to a watched resource.
type Event struct {
Type EventType
// If Type == Deleted, then this is the state of the object
// immediately before deletion.
Object interface{}
}
// FakeWatcher lets you test anything that consumes a watch.Interface; threadsafe.
type FakeWatcher struct {
result chan *Event
Stopped bool
sync.Mutex
}
func NewFake() *FakeWatcher {
return &FakeWatcher{
result: make(chan *Event),
}
}
// Stop implements Interface.Stop().
func (f *FakeWatcher) Stop() {
f.Lock()
defer f.Unlock()
close(f.result)
f.Stopped = true
}
func (f *FakeWatcher) ResultChan() <-chan *Event {
return f.result
}
// Add sends an add event.
func (f *FakeWatcher) Add(obj interface{}) {
f.result <- &Event{Added, obj}
}
// Modify sends a modify event.
func (f *FakeWatcher) Modify(obj interface{}) {
f.result <- &Event{Modified, obj}
}
// Delete sends a delete event.
func (f *FakeWatcher) Delete(lastValue interface{}) {
f.result <- &Event{Deleted, lastValue}
}
// Action sends an event of the requested type, for table-based testing.
func (f *FakeWatcher) Action(action EventType, obj interface{}) {
f.result <- &Event{action, obj}
}

66
pkg/watch/watch_test.go Normal file
View File

@ -0,0 +1,66 @@
/*
Copyright 2014 Google Inc. All rights reserved.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package watch
import (
"testing"
)
func TestFake(t *testing.T) {
f := NewFake()
table := []struct {
t EventType
s string
}{
{Added, "foo"},
{Modified, "qux"},
{Modified, "bar"},
{Deleted, "bar"},
}
// Prove that f implements Interface by phrasing this as a function.
consumer := func(w Interface) {
for _, expect := range table {
got, ok := <-w.ResultChan()
if !ok {
t.Fatalf("closed early")
}
if e, a := expect.t, got.Type; e != a {
t.Fatalf("Expected %v, got %v", e, a)
}
if a, ok := got.Object.(string); !ok || a != expect.s {
t.Fatalf("Expected %v, got %v", expect.s, a)
}
}
_, stillOpen := <-w.ResultChan()
if stillOpen {
t.Fatal("Never stopped")
}
}
sender := func() {
f.Add("foo")
f.Action(Modified, "qux")
f.Modify("bar")
f.Delete("bar")
f.Stop()
}
go sender()
consumer(f)
}

1
third_party/deps.sh vendored
View File

@ -15,6 +15,7 @@ DEP_PACKAGES="
code.google.com/p/google-api-go-client/googleapi
github.com/coreos/go-log/log
github.com/coreos/go-systemd/journal
code.google.com/p/go.net/websocket
"
PACKAGES="$TOP_PACKAGES $DEP_PACKAGES"

98
third_party/src/code.google.com/p/go.net/websocket/client.go vendored Normal file
View File

@ -0,0 +1,98 @@
// Copyright 2009 The Go Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
package websocket
import (
"bufio"
"crypto/tls"
"io"
"net"
"net/http"
"net/url"
)
// DialError is an error that occurs while dialling a websocket server.
type DialError struct {
*Config
Err error
}
func (e *DialError) Error() string {
return "websocket.Dial " + e.Config.Location.String() + ": " + e.Err.Error()
}
// NewConfig creates a new WebSocket config for client connection.
func NewConfig(server, origin string) (config *Config, err error) {
config = new(Config)
config.Version = ProtocolVersionHybi13
config.Location, err = url.ParseRequestURI(server)
if err != nil {
return
}
config.Origin, err = url.ParseRequestURI(origin)
if err != nil {
return
}
config.Header = http.Header(make(map[string][]string))
return
}
// NewClient creates a new WebSocket client connection over rwc.
func NewClient(config *Config, rwc io.ReadWriteCloser) (ws *Conn, err error) {
br := bufio.NewReader(rwc)
bw := bufio.NewWriter(rwc)
err = hybiClientHandshake(config, br, bw)
if err != nil {
return
}
buf := bufio.NewReadWriter(br, bw)
ws = newHybiClientConn(config, buf, rwc)
return
}
// Dial opens a new client connection to a WebSocket.
func Dial(url_, protocol, origin string) (ws *Conn, err error) {
config, err := NewConfig(url_, origin)
if err != nil {
return nil, err
}
if protocol != "" {
config.Protocol = []string{protocol}
}
return DialConfig(config)
}
// DialConfig opens a new client connection to a WebSocket with a config.
func DialConfig(config *Config) (ws *Conn, err error) {
var client net.Conn
if config.Location == nil {
return nil, &DialError{config, ErrBadWebSocketLocation}
}
if config.Origin == nil {
return nil, &DialError{config, ErrBadWebSocketOrigin}
}
switch config.Location.Scheme {
case "ws":
client, err = net.Dial("tcp", config.Location.Host)
case "wss":
client, err = tls.Dial("tcp", config.Location.Host, config.TlsConfig)
default:
err = ErrBadScheme
}
if err != nil {
goto Error
}
ws, err = NewClient(config, client)
if err != nil {
goto Error
}
return
Error:
return nil, &DialError{config, err}
}

31
third_party/src/code.google.com/p/go.net/websocket/exampledial_test.go vendored Normal file
View File

@ -0,0 +1,31 @@
// Copyright 2012 The Go Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
package websocket_test
import (
"fmt"
"log"
"code.google.com/p/go.net/websocket"
)
// This example demonstrates a trivial client.
func ExampleDial() {
origin := "http://localhost/"
url := "ws://localhost:12345/ws"
ws, err := websocket.Dial(url, "", origin)
if err != nil {
log.Fatal(err)
}
if _, err := ws.Write([]byte("hello, world!\n")); err != nil {
log.Fatal(err)
}
var msg = make([]byte, 512)
var n int
if n, err = ws.Read(msg); err != nil {
log.Fatal(err)
}
fmt.Printf("Received: %s.\n", msg[:n])
}

26
third_party/src/code.google.com/p/go.net/websocket/examplehandler_test.go vendored Normal file
View File

@ -0,0 +1,26 @@
// Copyright 2012 The Go Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
package websocket_test
import (
"io"
"net/http"
"code.google.com/p/go.net/websocket"
)
// Echo the data received on the WebSocket.
func EchoServer(ws *websocket.Conn) {
io.Copy(ws, ws)
}
// This example demonstrates a trivial echo server.
func ExampleHandler() {
http.Handle("/echo", websocket.Handler(EchoServer))
err := http.ListenAndServe(":12345", nil)
if err != nil {
panic("ListenAndServe: " + err.Error())
}
}

564
third_party/src/code.google.com/p/go.net/websocket/hybi.go vendored Normal file
View File

@ -0,0 +1,564 @@
// Copyright 2011 The Go Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
package websocket
// This file implements a protocol of hybi draft.
// http://tools.ietf.org/html/draft-ietf-hybi-thewebsocketprotocol-17
import (
"bufio"
"bytes"
"crypto/rand"
"crypto/sha1"
"encoding/base64"
"encoding/binary"
"fmt"
"io"
"io/ioutil"
"net/http"
"net/url"
"strings"
)
const (
websocketGUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"
closeStatusNormal = 1000
closeStatusGoingAway = 1001
closeStatusProtocolError = 1002
closeStatusUnsupportedData = 1003
closeStatusFrameTooLarge = 1004
closeStatusNoStatusRcvd = 1005
closeStatusAbnormalClosure = 1006
closeStatusBadMessageData = 1007
closeStatusPolicyViolation = 1008
closeStatusTooBigData = 1009
closeStatusExtensionMismatch = 1010
maxControlFramePayloadLength = 125
)
var (
ErrBadMaskingKey = &ProtocolError{"bad masking key"}
ErrBadPongMessage = &ProtocolError{"bad pong message"}
ErrBadClosingStatus = &ProtocolError{"bad closing status"}
ErrUnsupportedExtensions = &ProtocolError{"unsupported extensions"}
ErrNotImplemented = &ProtocolError{"not implemented"}
handshakeHeader = map[string]bool{
"Host": true,
"Upgrade": true,
"Connection": true,
"Sec-Websocket-Key": true,
"Sec-Websocket-Origin": true,
"Sec-Websocket-Version": true,
"Sec-Websocket-Protocol": true,
"Sec-Websocket-Accept": true,
}
)
// A hybiFrameHeader is a frame header as defined in hybi draft.
type hybiFrameHeader struct {
Fin bool
Rsv [3]bool
OpCode byte
Length int64
MaskingKey []byte
data *bytes.Buffer
}
// A hybiFrameReader is a reader for hybi frame.
type hybiFrameReader struct {
reader io.Reader
header hybiFrameHeader
pos int64
length int
}
func (frame *hybiFrameReader) Read(msg []byte) (n int, err error) {
n, err = frame.reader.Read(msg)
if err != nil {
return 0, err
}
if frame.header.MaskingKey != nil {
for i := 0; i < n; i++ {
msg[i] = msg[i] ^ frame.header.MaskingKey[frame.pos%4]
frame.pos++
}
}
return n, err
}
func (frame *hybiFrameReader) PayloadType() byte { return frame.header.OpCode }
func (frame *hybiFrameReader) HeaderReader() io.Reader {
if frame.header.data == nil {
return nil
}
if frame.header.data.Len() == 0 {
return nil
}
return frame.header.data
}
func (frame *hybiFrameReader) TrailerReader() io.Reader { return nil }
func (frame *hybiFrameReader) Len() (n int) { return frame.length }
// A hybiFrameReaderFactory creates new frame reader based on its frame type.
type hybiFrameReaderFactory struct {
*bufio.Reader
}
// NewFrameReader reads a frame header from the connection, and creates new reader for the frame.
// See Section 5.2 Base Framing protocol for detail.
// http://tools.ietf.org/html/draft-ietf-hybi-thewebsocketprotocol-17#section-5.2
func (buf hybiFrameReaderFactory) NewFrameReader() (frame frameReader, err error) {
hybiFrame := new(hybiFrameReader)
frame = hybiFrame
var header []byte
var b byte
// First byte. FIN/RSV1/RSV2/RSV3/OpCode(4bits)
b, err = buf.ReadByte()
if err != nil {
return
}
header = append(header, b)
hybiFrame.header.Fin = ((header[0] >> 7) & 1) != 0
for i := 0; i < 3; i++ {
j := uint(6 - i)
hybiFrame.header.Rsv[i] = ((header[0] >> j) & 1) != 0
}
hybiFrame.header.OpCode = header[0] & 0x0f
// Second byte. Mask/Payload len(7bits)
b, err = buf.ReadByte()
if err != nil {
return
}
header = append(header, b)
mask := (b & 0x80) != 0
b &= 0x7f
lengthFields := 0
switch {
case b <= 125: // Payload length 7bits.
hybiFrame.header.Length = int64(b)
case b == 126: // Payload length 7+16bits
lengthFields = 2
case b == 127: // Payload length 7+64bits
lengthFields = 8
}
for i := 0; i < lengthFields; i++ {
b, err = buf.ReadByte()
if err != nil {
return
}
header = append(header, b)
hybiFrame.header.Length = hybiFrame.header.Length*256 + int64(b)
}
if mask {
// Masking key. 4 bytes.
for i := 0; i < 4; i++ {
b, err = buf.ReadByte()
if err != nil {
return
}
header = append(header, b)
hybiFrame.header.MaskingKey = append(hybiFrame.header.MaskingKey, b)
}
}
hybiFrame.reader = io.LimitReader(buf.Reader, hybiFrame.header.Length)
hybiFrame.header.data = bytes.NewBuffer(header)
hybiFrame.length = len(header) + int(hybiFrame.header.Length)
return
}
// A HybiFrameWriter is a writer for hybi frame.
type hybiFrameWriter struct {
writer *bufio.Writer
header *hybiFrameHeader
}
func (frame *hybiFrameWriter) Write(msg []byte) (n int, err error) {
var header []byte
var b byte
if frame.header.Fin {
b |= 0x80
}
for i := 0; i < 3; i++ {
if frame.header.Rsv[i] {
j := uint(6 - i)
b |= 1 << j
}
}
b |= frame.header.OpCode
header = append(header, b)
if frame.header.MaskingKey != nil {
b = 0x80
} else {
b = 0
}
lengthFields := 0
length := len(msg)
switch {
case length <= 125:
b |= byte(length)
case length < 65536:
b |= 126
lengthFields = 2
default:
b |= 127
lengthFields = 8
}
header = append(header, b)
for i := 0; i < lengthFields; i++ {
j := uint((lengthFields - i - 1) * 8)
b = byte((length >> j) & 0xff)
header = append(header, b)
}
if frame.header.MaskingKey != nil {
if len(frame.header.MaskingKey) != 4 {
return 0, ErrBadMaskingKey
}
header = append(header, frame.header.MaskingKey...)
frame.writer.Write(header)
data := make([]byte, length)
for i := range data {
data[i] = msg[i] ^ frame.header.MaskingKey[i%4]
}
frame.writer.Write(data)
err = frame.writer.Flush()
return length, err
}
frame.writer.Write(header)
frame.writer.Write(msg)
err = frame.writer.Flush()
return length, err
}
func (frame *hybiFrameWriter) Close() error { return nil }
type hybiFrameWriterFactory struct {
*bufio.Writer
needMaskingKey bool
}
func (buf hybiFrameWriterFactory) NewFrameWriter(payloadType byte) (frame frameWriter, err error) {
frameHeader := &hybiFrameHeader{Fin: true, OpCode: payloadType}
if buf.needMaskingKey {
frameHeader.MaskingKey, err = generateMaskingKey()
if err != nil {
return nil, err
}
}
return &hybiFrameWriter{writer: buf.Writer, header: frameHeader}, nil
}
type hybiFrameHandler struct {
conn *Conn
payloadType byte
}
func (handler *hybiFrameHandler) HandleFrame(frame frameReader) (r frameReader, err error) {
if handler.conn.IsServerConn() {
// The client MUST mask all frames sent to the server.
if frame.(*hybiFrameReader).header.MaskingKey == nil {
handler.WriteClose(closeStatusProtocolError)
return nil, io.EOF
}
} else {
// The server MUST NOT mask all frames.
if frame.(*hybiFrameReader).header.MaskingKey != nil {
handler.WriteClose(closeStatusProtocolError)
return nil, io.EOF
}
}
if header := frame.HeaderReader(); header != nil {
io.Copy(ioutil.Discard, header)
}
switch frame.PayloadType() {
case ContinuationFrame:
frame.(*hybiFrameReader).header.OpCode = handler.payloadType
case TextFrame, BinaryFrame:
handler.payloadType = frame.PayloadType()
case CloseFrame:
return nil, io.EOF
case PingFrame:
pingMsg := make([]byte, maxControlFramePayloadLength)
n, err := io.ReadFull(frame, pingMsg)
if err != nil && err != io.ErrUnexpectedEOF {
return nil, err
}
io.Copy(ioutil.Discard, frame)
n, err = handler.WritePong(pingMsg[:n])
if err != nil {
return nil, err
}
return nil, nil
case PongFrame:
return nil, ErrNotImplemented
}
return frame, nil
}
func (handler *hybiFrameHandler) WriteClose(status int) (err error) {
handler.conn.wio.Lock()
defer handler.conn.wio.Unlock()
w, err := handler.conn.frameWriterFactory.NewFrameWriter(CloseFrame)
if err != nil {
return err
}
msg := make([]byte, 2)
binary.BigEndian.PutUint16(msg, uint16(status))
_, err = w.Write(msg)
w.Close()
return err
}
func (handler *hybiFrameHandler) WritePong(msg []byte) (n int, err error) {
handler.conn.wio.Lock()
defer handler.conn.wio.Unlock()
w, err := handler.conn.frameWriterFactory.NewFrameWriter(PongFrame)
if err != nil {
return 0, err
}
n, err = w.Write(msg)
w.Close()
return n, err
}
// newHybiConn creates a new WebSocket connection speaking hybi draft protocol.
func newHybiConn(config *Config, buf *bufio.ReadWriter, rwc io.ReadWriteCloser, request *http.Request) *Conn {
if buf == nil {
br := bufio.NewReader(rwc)
bw := bufio.NewWriter(rwc)
buf = bufio.NewReadWriter(br, bw)
}
ws := &Conn{config: config, request: request, buf: buf, rwc: rwc,
frameReaderFactory: hybiFrameReaderFactory{buf.Reader},
frameWriterFactory: hybiFrameWriterFactory{
buf.Writer, request == nil},
PayloadType: TextFrame,
defaultCloseStatus: closeStatusNormal}
ws.frameHandler = &hybiFrameHandler{conn: ws}
return ws
}
// generateMaskingKey generates a masking key for a frame.
func generateMaskingKey() (maskingKey []byte, err error) {
maskingKey = make([]byte, 4)
if _, err = io.ReadFull(rand.Reader, maskingKey); err != nil {
return
}
return
}
// generateNonce generates a nonce consisting of a randomly selected 16-byte
// value that has been base64-encoded.
func generateNonce() (nonce []byte) {
key := make([]byte, 16)
if _, err := io.ReadFull(rand.Reader, key); err != nil {
panic(err)
}
nonce = make([]byte, 24)
base64.StdEncoding.Encode(nonce, key)
return
}
// getNonceAccept computes the base64-encoded SHA-1 of the concatenation of
// the nonce ("Sec-WebSocket-Key" value) with the websocket GUID string.
func getNonceAccept(nonce []byte) (expected []byte, err error) {
h := sha1.New()
if _, err = h.Write(nonce); err != nil {
return
}
if _, err = h.Write([]byte(websocketGUID)); err != nil {
return
}
expected = make([]byte, 28)
base64.StdEncoding.Encode(expected, h.Sum(nil))
return
}
// Client handshake described in draft-ietf-hybi-thewebsocket-protocol-17
func hybiClientHandshake(config *Config, br *bufio.Reader, bw *bufio.Writer) (err error) {
bw.WriteString("GET " + config.Location.RequestURI() + " HTTP/1.1\r\n")
bw.WriteString("Host: " + config.Location.Host + "\r\n")
bw.WriteString("Upgrade: websocket\r\n")
bw.WriteString("Connection: Upgrade\r\n")
nonce := generateNonce()
if config.handshakeData != nil {
nonce = []byte(config.handshakeData["key"])
}
bw.WriteString("Sec-WebSocket-Key: " + string(nonce) + "\r\n")
bw.WriteString("Origin: " + strings.ToLower(config.Origin.String()) + "\r\n")
if config.Version != ProtocolVersionHybi13 {
return ErrBadProtocolVersion
}
bw.WriteString("Sec-WebSocket-Version: " + fmt.Sprintf("%d", config.Version) + "\r\n")
if len(config.Protocol) > 0 {
bw.WriteString("Sec-WebSocket-Protocol: " + strings.Join(config.Protocol, ", ") + "\r\n")
}
// TODO(ukai): send Sec-WebSocket-Extensions.
err = config.Header.WriteSubset(bw, handshakeHeader)
if err != nil {
return err
}
bw.WriteString("\r\n")
if err = bw.Flush(); err != nil {
return err
}
resp, err := http.ReadResponse(br, &http.Request{Method: "GET"})
if err != nil {
return err
}
if resp.StatusCode != 101 {
return ErrBadStatus
}
if strings.ToLower(resp.Header.Get("Upgrade")) != "websocket" ||
strings.ToLower(resp.Header.Get("Connection")) != "upgrade" {
return ErrBadUpgrade
}
expectedAccept, err := getNonceAccept(nonce)
if err != nil {
return err
}
if resp.Header.Get("Sec-WebSocket-Accept") != string(expectedAccept) {
return ErrChallengeResponse
}
if resp.Header.Get("Sec-WebSocket-Extensions") != "" {
return ErrUnsupportedExtensions
}
offeredProtocol := resp.Header.Get("Sec-WebSocket-Protocol")
if offeredProtocol != "" {
protocolMatched := false
for i := 0; i < len(config.Protocol); i++ {
if config.Protocol[i] == offeredProtocol {
protocolMatched = true
break
}
}
if !protocolMatched {
return ErrBadWebSocketProtocol
}
config.Protocol = []string{offeredProtocol}
}
return nil
}
// newHybiClientConn creates a client WebSocket connection after handshake.
func newHybiClientConn(config *Config, buf *bufio.ReadWriter, rwc io.ReadWriteCloser) *Conn {
return newHybiConn(config, buf, rwc, nil)
}
// A HybiServerHandshaker performs a server handshake using hybi draft protocol.
type hybiServerHandshaker struct {
*Config
accept []byte
}
func (c *hybiServerHandshaker) ReadHandshake(buf *bufio.Reader, req *http.Request) (code int, err error) {
c.Version = ProtocolVersionHybi13
if req.Method != "GET" {
return http.StatusMethodNotAllowed, ErrBadRequestMethod
}
// HTTP version can be safely ignored.
if strings.ToLower(req.Header.Get("Upgrade")) != "websocket" ||
!strings.Contains(strings.ToLower(req.Header.Get("Connection")), "upgrade") {
return http.StatusBadRequest, ErrNotWebSocket
}
key := req.Header.Get("Sec-Websocket-Key")
if key == "" {
return http.StatusBadRequest, ErrChallengeResponse
}
version := req.Header.Get("Sec-Websocket-Version")
switch version {
case "13":
c.Version = ProtocolVersionHybi13
default:
return http.StatusBadRequest, ErrBadWebSocketVersion
}
var scheme string
if req.TLS != nil {
scheme = "wss"
} else {
scheme = "ws"
}
c.Location, err = url.ParseRequestURI(scheme + "://" + req.Host + req.URL.RequestURI())
if err != nil {
return http.StatusBadRequest, err
}
protocol := strings.TrimSpace(req.Header.Get("Sec-Websocket-Protocol"))
if protocol != "" {
protocols := strings.Split(protocol, ",")
for i := 0; i < len(protocols); i++ {
c.Protocol = append(c.Protocol, strings.TrimSpace(protocols[i]))
}
}
c.accept, err = getNonceAccept([]byte(key))
if err != nil {
return http.StatusInternalServerError, err
}
return http.StatusSwitchingProtocols, nil
}
// Origin parses Origin header in "req".
// If origin is "null", returns (nil, nil).
func Origin(config *Config, req *http.Request) (*url.URL, error) {
var origin string
switch config.Version {
case ProtocolVersionHybi13:
origin = req.Header.Get("Origin")
}
if origin == "null" {
return nil, nil
}
return url.ParseRequestURI(origin)
}
func (c *hybiServerHandshaker) AcceptHandshake(buf *bufio.Writer) (err error) {
if len(c.Protocol) > 0 {
if len(c.Protocol) != 1 {
// You need choose a Protocol in Handshake func in Server.
return ErrBadWebSocketProtocol
}
}
buf.WriteString("HTTP/1.1 101 Switching Protocols\r\n")
buf.WriteString("Upgrade: websocket\r\n")
buf.WriteString("Connection: Upgrade\r\n")
buf.WriteString("Sec-WebSocket-Accept: " + string(c.accept) + "\r\n")
if len(c.Protocol) > 0 {
buf.WriteString("Sec-WebSocket-Protocol: " + c.Protocol[0] + "\r\n")
}
// TODO(ukai): send Sec-WebSocket-Extensions.
if c.Header != nil {
err := c.Header.WriteSubset(buf, handshakeHeader)
if err != nil {
return err
}
}
buf.WriteString("\r\n")
return buf.Flush()
}
func (c *hybiServerHandshaker) NewServerConn(buf *bufio.ReadWriter, rwc io.ReadWriteCloser, request *http.Request) *Conn {
return newHybiServerConn(c.Config, buf, rwc, request)
}
// newHybiServerConn returns a new WebSocket connection speaking hybi draft protocol.
func newHybiServerConn(config *Config, buf *bufio.ReadWriter, rwc io.ReadWriteCloser, request *http.Request) *Conn {
return newHybiConn(config, buf, rwc, request)
}

590
third_party/src/code.google.com/p/go.net/websocket/hybi_test.go vendored Normal file
View File

@ -0,0 +1,590 @@
// Copyright 2011 The Go Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
package websocket
import (
"bufio"
"bytes"
"fmt"
"io"
"net/http"
"net/url"
"strings"
"testing"
)
// Test the getNonceAccept function with values in
// http://tools.ietf.org/html/draft-ietf-hybi-thewebsocketprotocol-17
func TestSecWebSocketAccept(t *testing.T) {
nonce := []byte("dGhlIHNhbXBsZSBub25jZQ==")
expected := []byte("s3pPLMBiTxaQ9kYGzzhZRbK+xOo=")
accept, err := getNonceAccept(nonce)
if err != nil {
t.Errorf("getNonceAccept: returned error %v", err)
return
}
if !bytes.Equal(expected, accept) {
t.Errorf("getNonceAccept: expected %q got %q", expected, accept)
}
}
func TestHybiClientHandshake(t *testing.T) {
b := bytes.NewBuffer([]byte{})
bw := bufio.NewWriter(b)
br := bufio.NewReader(strings.NewReader(`HTTP/1.1 101 Switching Protocols
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=
Sec-WebSocket-Protocol: chat
`))
var err error
config := new(Config)
config.Location, err = url.ParseRequestURI("ws://server.example.com/chat")
if err != nil {
t.Fatal("location url", err)
}
config.Origin, err = url.ParseRequestURI("http://example.com")
if err != nil {
t.Fatal("origin url", err)
}
config.Protocol = append(config.Protocol, "chat")
config.Protocol = append(config.Protocol, "superchat")
config.Version = ProtocolVersionHybi13
config.handshakeData = map[string]string{
"key": "dGhlIHNhbXBsZSBub25jZQ==",
}
err = hybiClientHandshake(config, br, bw)
if err != nil {
t.Errorf("handshake failed: %v", err)
}
req, err := http.ReadRequest(bufio.NewReader(b))
if err != nil {
t.Fatalf("read request: %v", err)
}
if req.Method != "GET" {
t.Errorf("request method expected GET, but got %q", req.Method)
}
if req.URL.Path != "/chat" {
t.Errorf("request path expected /chat, but got %q", req.URL.Path)
}
if req.Proto != "HTTP/1.1" {
t.Errorf("request proto expected HTTP/1.1, but got %q", req.Proto)
}
if req.Host != "server.example.com" {
t.Errorf("request Host expected server.example.com, but got %v", req.Host)
}
var expectedHeader = map[string]string{
"Connection": "Upgrade",
"Upgrade": "websocket",
"Sec-Websocket-Key": config.handshakeData["key"],
"Origin": config.Origin.String(),
"Sec-Websocket-Protocol": "chat, superchat",
"Sec-Websocket-Version": fmt.Sprintf("%d", ProtocolVersionHybi13),
}
for k, v := range expectedHeader {
if req.Header.Get(k) != v {
t.Errorf(fmt.Sprintf("%s expected %q but got %q", k, v, req.Header.Get(k)))
}
}
}
func TestHybiClientHandshakeWithHeader(t *testing.T) {
b := bytes.NewBuffer([]byte{})
bw := bufio.NewWriter(b)
br := bufio.NewReader(strings.NewReader(`HTTP/1.1 101 Switching Protocols
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=
Sec-WebSocket-Protocol: chat
`))
var err error
config := new(Config)
config.Location, err = url.ParseRequestURI("ws://server.example.com/chat")
if err != nil {
t.Fatal("location url", err)
}
config.Origin, err = url.ParseRequestURI("http://example.com")
if err != nil {
t.Fatal("origin url", err)
}
config.Protocol = append(config.Protocol, "chat")
config.Protocol = append(config.Protocol, "superchat")
config.Version = ProtocolVersionHybi13
config.Header = http.Header(make(map[string][]string))
config.Header.Add("User-Agent", "test")
config.handshakeData = map[string]string{
"key": "dGhlIHNhbXBsZSBub25jZQ==",
}
err = hybiClientHandshake(config, br, bw)
if err != nil {
t.Errorf("handshake failed: %v", err)
}
req, err := http.ReadRequest(bufio.NewReader(b))
if err != nil {
t.Fatalf("read request: %v", err)
}
if req.Method != "GET" {
t.Errorf("request method expected GET, but got %q", req.Method)
}
if req.URL.Path != "/chat" {
t.Errorf("request path expected /chat, but got %q", req.URL.Path)
}
if req.Proto != "HTTP/1.1" {
t.Errorf("request proto expected HTTP/1.1, but got %q", req.Proto)
}
if req.Host != "server.example.com" {
t.Errorf("request Host expected server.example.com, but got %v", req.Host)
}
var expectedHeader = map[string]string{
"Connection": "Upgrade",
"Upgrade": "websocket",
"Sec-Websocket-Key": config.handshakeData["key"],
"Origin": config.Origin.String(),
"Sec-Websocket-Protocol": "chat, superchat",
"Sec-Websocket-Version": fmt.Sprintf("%d", ProtocolVersionHybi13),
"User-Agent": "test",
}
for k, v := range expectedHeader {
if req.Header.Get(k) != v {
t.Errorf(fmt.Sprintf("%s expected %q but got %q", k, v, req.Header.Get(k)))
}
}
}
func TestHybiServerHandshake(t *testing.T) {
config := new(Config)
handshaker := &hybiServerHandshaker{Config: config}
br := bufio.NewReader(strings.NewReader(`GET /chat HTTP/1.1
Host: server.example.com
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==
Origin: http://example.com
Sec-WebSocket-Protocol: chat, superchat
Sec-WebSocket-Version: 13
`))
req, err := http.ReadRequest(br)
if err != nil {
t.Fatal("request", err)
}
code, err := handshaker.ReadHandshake(br, req)
if err != nil {
t.Errorf("handshake failed: %v", err)
}
if code != http.StatusSwitchingProtocols {
t.Errorf("status expected %q but got %q", http.StatusSwitchingProtocols, code)
}
expectedProtocols := []string{"chat", "superchat"}
if fmt.Sprintf("%v", config.Protocol) != fmt.Sprintf("%v", expectedProtocols) {
t.Errorf("protocol expected %q but got %q", expectedProtocols, config.Protocol)
}
b := bytes.NewBuffer([]byte{})
bw := bufio.NewWriter(b)
config.Protocol = config.Protocol[:1]
err = handshaker.AcceptHandshake(bw)
if err != nil {
t.Errorf("handshake response failed: %v", err)
}
expectedResponse := strings.Join([]string{
"HTTP/1.1 101 Switching Protocols",
"Upgrade: websocket",
"Connection: Upgrade",
"Sec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=",
"Sec-WebSocket-Protocol: chat",
"", ""}, "\r\n")
if b.String() != expectedResponse {
t.Errorf("handshake expected %q but got %q", expectedResponse, b.String())
}
}
func TestHybiServerHandshakeNoSubProtocol(t *testing.T) {
config := new(Config)
handshaker := &hybiServerHandshaker{Config: config}
br := bufio.NewReader(strings.NewReader(`GET /chat HTTP/1.1
Host: server.example.com
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==
Origin: http://example.com
Sec-WebSocket-Version: 13
`))
req, err := http.ReadRequest(br)
if err != nil {
t.Fatal("request", err)
}
code, err := handshaker.ReadHandshake(br, req)
if err != nil {
t.Errorf("handshake failed: %v", err)
}
if code != http.StatusSwitchingProtocols {
t.Errorf("status expected %q but got %q", http.StatusSwitchingProtocols, code)
}
if len(config.Protocol) != 0 {
t.Errorf("len(config.Protocol) expected 0, but got %q", len(config.Protocol))
}
b := bytes.NewBuffer([]byte{})
bw := bufio.NewWriter(b)
err = handshaker.AcceptHandshake(bw)
if err != nil {
t.Errorf("handshake response failed: %v", err)
}
expectedResponse := strings.Join([]string{
"HTTP/1.1 101 Switching Protocols",
"Upgrade: websocket",
"Connection: Upgrade",
"Sec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=",
"", ""}, "\r\n")
if b.String() != expectedResponse {
t.Errorf("handshake expected %q but got %q", expectedResponse, b.String())
}
}
func TestHybiServerHandshakeHybiBadVersion(t *testing.T) {
config := new(Config)
handshaker := &hybiServerHandshaker{Config: config}
br := bufio.NewReader(strings.NewReader(`GET /chat HTTP/1.1
Host: server.example.com
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==
Sec-WebSocket-Origin: http://example.com
Sec-WebSocket-Protocol: chat, superchat
Sec-WebSocket-Version: 9
`))
req, err := http.ReadRequest(br)
if err != nil {
t.Fatal("request", err)
}
code, err := handshaker.ReadHandshake(br, req)
if err != ErrBadWebSocketVersion {
t.Errorf("handshake expected err %q but got %q", ErrBadWebSocketVersion, err)
}
if code != http.StatusBadRequest {
t.Errorf("status expected %q but got %q", http.StatusBadRequest, code)
}
}
func testHybiFrame(t *testing.T, testHeader, testPayload, testMaskedPayload []byte, frameHeader *hybiFrameHeader) {
b := bytes.NewBuffer([]byte{})
frameWriterFactory := &hybiFrameWriterFactory{bufio.NewWriter(b), false}
w, _ := frameWriterFactory.NewFrameWriter(TextFrame)
w.(*hybiFrameWriter).header = frameHeader
_, err := w.Write(testPayload)
w.Close()
if err != nil {
t.Errorf("Write error %q", err)
}
var expectedFrame []byte
expectedFrame = append(expectedFrame, testHeader...)
expectedFrame = append(expectedFrame, testMaskedPayload...)
if !bytes.Equal(expectedFrame, b.Bytes()) {
t.Errorf("frame expected %q got %q", expectedFrame, b.Bytes())
}
frameReaderFactory := &hybiFrameReaderFactory{bufio.NewReader(b)}
r, err := frameReaderFactory.NewFrameReader()
if err != nil {
t.Errorf("Read error %q", err)
}
if header := r.HeaderReader(); header == nil {
t.Errorf("no header")
} else {
actualHeader := make([]byte, r.Len())
n, err := header.Read(actualHeader)
if err != nil {
t.Errorf("Read header error %q", err)
} else {
if n < len(testHeader) {
t.Errorf("header too short %q got %q", testHeader, actualHeader[:n])
}
if !bytes.Equal(testHeader, actualHeader[:n]) {
t.Errorf("header expected %q got %q", testHeader, actualHeader[:n])
}
}
}
if trailer := r.TrailerReader(); trailer != nil {
t.Errorf("unexpected trailer %q", trailer)
}
frame := r.(*hybiFrameReader)
if frameHeader.Fin != frame.header.Fin ||
frameHeader.OpCode != frame.header.OpCode ||
len(testPayload) != int(frame.header.Length) {
t.Errorf("mismatch %v (%d) vs %v", frameHeader, len(testPayload), frame)
}
payload := make([]byte, len(testPayload))
_, err = r.Read(payload)
if err != nil {
t.Errorf("read %v", err)
}
if !bytes.Equal(testPayload, payload) {
t.Errorf("payload %q vs %q", testPayload, payload)
}
}
func TestHybiShortTextFrame(t *testing.T) {
frameHeader := &hybiFrameHeader{Fin: true, OpCode: TextFrame}
payload := []byte("hello")
testHybiFrame(t, []byte{0x81, 0x05}, payload, payload, frameHeader)
payload = make([]byte, 125)
testHybiFrame(t, []byte{0x81, 125}, payload, payload, frameHeader)
}
func TestHybiShortMaskedTextFrame(t *testing.T) {
frameHeader := &hybiFrameHeader{Fin: true, OpCode: TextFrame,
MaskingKey: []byte{0xcc, 0x55, 0x80, 0x20}}
payload := []byte("hello")
maskedPayload := []byte{0xa4, 0x30, 0xec, 0x4c, 0xa3}
header := []byte{0x81, 0x85}
header = append(header, frameHeader.MaskingKey...)
testHybiFrame(t, header, payload, maskedPayload, frameHeader)
}
func TestHybiShortBinaryFrame(t *testing.T) {
frameHeader := &hybiFrameHeader{Fin: true, OpCode: BinaryFrame}
payload := []byte("hello")
testHybiFrame(t, []byte{0x82, 0x05}, payload, payload, frameHeader)
payload = make([]byte, 125)
testHybiFrame(t, []byte{0x82, 125}, payload, payload, frameHeader)
}
func TestHybiControlFrame(t *testing.T) {
frameHeader := &hybiFrameHeader{Fin: true, OpCode: PingFrame}
payload := []byte("hello")
testHybiFrame(t, []byte{0x89, 0x05}, payload, payload, frameHeader)
frameHeader = &hybiFrameHeader{Fin: true, OpCode: PongFrame}
testHybiFrame(t, []byte{0x8A, 0x05}, payload, payload, frameHeader)
frameHeader = &hybiFrameHeader{Fin: true, OpCode: CloseFrame}
payload = []byte{0x03, 0xe8} // 1000
testHybiFrame(t, []byte{0x88, 0x02}, payload, payload, frameHeader)
}
func TestHybiLongFrame(t *testing.T) {
frameHeader := &hybiFrameHeader{Fin: true, OpCode: TextFrame}
payload := make([]byte, 126)
testHybiFrame(t, []byte{0x81, 126, 0x00, 126}, payload, payload, frameHeader)
payload = make([]byte, 65535)
testHybiFrame(t, []byte{0x81, 126, 0xff, 0xff}, payload, payload, frameHeader)
payload = make([]byte, 65536)
testHybiFrame(t, []byte{0x81, 127, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01, 0x00, 0x00}, payload, payload, frameHeader)
}
func TestHybiClientRead(t *testing.T) {
wireData := []byte{0x81, 0x05, 'h', 'e', 'l', 'l', 'o',
0x89, 0x05, 'h', 'e', 'l', 'l', 'o', // ping
0x81, 0x05, 'w', 'o', 'r', 'l', 'd'}
br := bufio.NewReader(bytes.NewBuffer(wireData))
bw := bufio.NewWriter(bytes.NewBuffer([]byte{}))
conn := newHybiConn(newConfig(t, "/"), bufio.NewReadWriter(br, bw), nil, nil)
msg := make([]byte, 512)
n, err := conn.Read(msg)
if err != nil {
t.Errorf("read 1st frame, error %q", err)
}
if n != 5 {
t.Errorf("read 1st frame, expect 5, got %d", n)
}
if !bytes.Equal(wireData[2:7], msg[:n]) {
t.Errorf("read 1st frame %v, got %v", wireData[2:7], msg[:n])
}
n, err = conn.Read(msg)
if err != nil {
t.Errorf("read 2nd frame, error %q", err)
}
if n != 5 {
t.Errorf("read 2nd frame, expect 5, got %d", n)
}
if !bytes.Equal(wireData[16:21], msg[:n]) {
t.Errorf("read 2nd frame %v, got %v", wireData[16:21], msg[:n])
}
n, err = conn.Read(msg)
if err == nil {
t.Errorf("read not EOF")
}
if n != 0 {
t.Errorf("expect read 0, got %d", n)
}
}
func TestHybiShortRead(t *testing.T) {
wireData := []byte{0x81, 0x05, 'h', 'e', 'l', 'l', 'o',
0x89, 0x05, 'h', 'e', 'l', 'l', 'o', // ping
0x81, 0x05, 'w', 'o', 'r', 'l', 'd'}
br := bufio.NewReader(bytes.NewBuffer(wireData))
bw := bufio.NewWriter(bytes.NewBuffer([]byte{}))
conn := newHybiConn(newConfig(t, "/"), bufio.NewReadWriter(br, bw), nil, nil)
step := 0
pos := 0
expectedPos := []int{2, 5, 16, 19}
expectedLen := []int{3, 2, 3, 2}
for {
msg := make([]byte, 3)
n, err := conn.Read(msg)
if step >= len(expectedPos) {
if err == nil {
t.Errorf("read not EOF")
}
if n != 0 {
t.Errorf("expect read 0, got %d", n)
}
return
}
pos = expectedPos[step]
endPos := pos + expectedLen[step]
if err != nil {
t.Errorf("read from %d, got error %q", pos, err)
return
}
if n != endPos-pos {
t.Errorf("read from %d, expect %d, got %d", pos, endPos-pos, n)
}
if !bytes.Equal(wireData[pos:endPos], msg[:n]) {
t.Errorf("read from %d, frame %v, got %v", pos, wireData[pos:endPos], msg[:n])
}
step++
}
}
func TestHybiServerRead(t *testing.T) {
wireData := []byte{0x81, 0x85, 0xcc, 0x55, 0x80, 0x20,
0xa4, 0x30, 0xec, 0x4c, 0xa3, // hello
0x89, 0x85, 0xcc, 0x55, 0x80, 0x20,
0xa4, 0x30, 0xec, 0x4c, 0xa3, // ping: hello
0x81, 0x85, 0xed, 0x83, 0xb4, 0x24,
0x9a, 0xec, 0xc6, 0x48, 0x89, // world
}
br := bufio.NewReader(bytes.NewBuffer(wireData))
bw := bufio.NewWriter(bytes.NewBuffer([]byte{}))
conn := newHybiConn(newConfig(t, "/"), bufio.NewReadWriter(br, bw), nil, new(http.Request))
expected := [][]byte{[]byte("hello"), []byte("world")}
msg := make([]byte, 512)
n, err := conn.Read(msg)
if err != nil {
t.Errorf("read 1st frame, error %q", err)
}
if n != 5 {
t.Errorf("read 1st frame, expect 5, got %d", n)
}
if !bytes.Equal(expected[0], msg[:n]) {
t.Errorf("read 1st frame %q, got %q", expected[0], msg[:n])
}
n, err = conn.Read(msg)
if err != nil {
t.Errorf("read 2nd frame, error %q", err)
}
if n != 5 {
t.Errorf("read 2nd frame, expect 5, got %d", n)
}
if !bytes.Equal(expected[1], msg[:n]) {
t.Errorf("read 2nd frame %q, got %q", expected[1], msg[:n])
}
n, err = conn.Read(msg)
if err == nil {
t.Errorf("read not EOF")
}
if n != 0 {
t.Errorf("expect read 0, got %d", n)
}
}
func TestHybiServerReadWithoutMasking(t *testing.T) {
wireData := []byte{0x81, 0x05, 'h', 'e', 'l', 'l', 'o'}
br := bufio.NewReader(bytes.NewBuffer(wireData))
bw := bufio.NewWriter(bytes.NewBuffer([]byte{}))
conn := newHybiConn(newConfig(t, "/"), bufio.NewReadWriter(br, bw), nil, new(http.Request))
// server MUST close the connection upon receiving a non-masked frame.
msg := make([]byte, 512)
_, err := conn.Read(msg)
if err != io.EOF {
t.Errorf("read 1st frame, expect %q, but got %q", io.EOF, err)
}
}
func TestHybiClientReadWithMasking(t *testing.T) {
wireData := []byte{0x81, 0x85, 0xcc, 0x55, 0x80, 0x20,
0xa4, 0x30, 0xec, 0x4c, 0xa3, // hello
}
br := bufio.NewReader(bytes.NewBuffer(wireData))
bw := bufio.NewWriter(bytes.NewBuffer([]byte{}))
conn := newHybiConn(newConfig(t, "/"), bufio.NewReadWriter(br, bw), nil, nil)
// client MUST close the connection upon receiving a masked frame.
msg := make([]byte, 512)
_, err := conn.Read(msg)
if err != io.EOF {
t.Errorf("read 1st frame, expect %q, but got %q", io.EOF, err)
}
}
// Test the hybiServerHandshaker supports firefox implementation and
// checks Connection request header include (but it's not necessary
// equal to) "upgrade"
func TestHybiServerFirefoxHandshake(t *testing.T) {
config := new(Config)
handshaker := &hybiServerHandshaker{Config: config}
br := bufio.NewReader(strings.NewReader(`GET /chat HTTP/1.1
Host: server.example.com
Upgrade: websocket
Connection: keep-alive, upgrade
Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==
Origin: http://example.com
Sec-WebSocket-Protocol: chat, superchat
Sec-WebSocket-Version: 13
`))
req, err := http.ReadRequest(br)
if err != nil {
t.Fatal("request", err)
}
code, err := handshaker.ReadHandshake(br, req)
if err != nil {
t.Errorf("handshake failed: %v", err)
}
if code != http.StatusSwitchingProtocols {
t.Errorf("status expected %q but got %q", http.StatusSwitchingProtocols, code)
}
b := bytes.NewBuffer([]byte{})
bw := bufio.NewWriter(b)
config.Protocol = []string{"chat"}
err = handshaker.AcceptHandshake(bw)
if err != nil {
t.Errorf("handshake response failed: %v", err)
}
expectedResponse := strings.Join([]string{
"HTTP/1.1 101 Switching Protocols",
"Upgrade: websocket",
"Connection: Upgrade",
"Sec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=",
"Sec-WebSocket-Protocol: chat",
"", ""}, "\r\n")
if b.String() != expectedResponse {
t.Errorf("handshake expected %q but got %q", expectedResponse, b.String())
}
}

114
third_party/src/code.google.com/p/go.net/websocket/server.go vendored Normal file
View File

@ -0,0 +1,114 @@
// Copyright 2009 The Go Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
package websocket
import (
"bufio"
"fmt"
"io"
"net/http"
)
func newServerConn(rwc io.ReadWriteCloser, buf *bufio.ReadWriter, req *http.Request, config *Config, handshake func(*Config, *http.Request) error) (conn *Conn, err error) {
var hs serverHandshaker = &hybiServerHandshaker{Config: config}
code, err := hs.ReadHandshake(buf.Reader, req)
if err == ErrBadWebSocketVersion {
fmt.Fprintf(buf, "HTTP/1.1 %03d %s\r\n", code, http.StatusText(code))
fmt.Fprintf(buf, "Sec-WebSocket-Version: %s\r\n", SupportedProtocolVersion)
buf.WriteString("\r\n")
buf.WriteString(err.Error())
buf.Flush()
return
}
if err != nil {
fmt.Fprintf(buf, "HTTP/1.1 %03d %s\r\n", code, http.StatusText(code))
buf.WriteString("\r\n")
buf.WriteString(err.Error())
buf.Flush()
return
}
if handshake != nil {
err = handshake(config, req)
if err != nil {
code = http.StatusForbidden
fmt.Fprintf(buf, "HTTP/1.1 %03d %s\r\n", code, http.StatusText(code))
buf.WriteString("\r\n")
buf.Flush()
return
}
}
err = hs.AcceptHandshake(buf.Writer)
if err != nil {
code = http.StatusBadRequest
fmt.Fprintf(buf, "HTTP/1.1 %03d %s\r\n", code, http.StatusText(code))
buf.WriteString("\r\n")
buf.Flush()
return
}
conn = hs.NewServerConn(buf, rwc, req)
return
}
// Server represents a server of a WebSocket.
type Server struct {
// Config is a WebSocket configuration for new WebSocket connection.
Config
// Handshake is an optional function in WebSocket handshake.
// For example, you can check, or don't check Origin header.
// Another example, you can select config.Protocol.
Handshake func(*Config, *http.Request) error
// Handler handles a WebSocket connection.
Handler
}
// ServeHTTP implements the http.Handler interface for a WebSocket
func (s Server) ServeHTTP(w http.ResponseWriter, req *http.Request) {
s.serveWebSocket(w, req)
}
func (s Server) serveWebSocket(w http.ResponseWriter, req *http.Request) {
rwc, buf, err := w.(http.Hijacker).Hijack()
if err != nil {
panic("Hijack failed: " + err.Error())
return
}
// The server should abort the WebSocket connection if it finds
// the client did not send a handshake that matches with protocol
// specification.
defer rwc.Close()
conn, err := newServerConn(rwc, buf, req, &s.Config, s.Handshake)
if err != nil {
return
}
if conn == nil {
panic("unexpected nil conn")
}
s.Handler(conn)
}
// Handler is a simple interface to a WebSocket browser client.
// It checks if Origin header is valid URL by default.
// You might want to verify websocket.Conn.Config().Origin in the func.
// If you use Server instead of Handler, you could call websocket.Origin and
// check the origin in your Handshake func. So, if you want to accept
// non-browser client, which doesn't send Origin header, you could use Server
//. that doesn't check origin in its Handshake.
type Handler func(*Conn)
func checkOrigin(config *Config, req *http.Request) (err error) {
config.Origin, err = Origin(config, req)
if err == nil && config.Origin == nil {
return fmt.Errorf("null origin")
}
return err
}
// ServeHTTP implements the http.Handler interface for a WebSocket
func (h Handler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
s := Server{Handler: h, Handshake: checkOrigin}
s.serveWebSocket(w, req)
}

411
third_party/src/code.google.com/p/go.net/websocket/websocket.go vendored Normal file
View File

@ -0,0 +1,411 @@
// Copyright 2009 The Go Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
// Package websocket implements a client and server for the WebSocket protocol
// as specified in RFC 6455.
package websocket
import (
"bufio"
"crypto/tls"
"encoding/json"
"errors"
"io"
"io/ioutil"
"net"
"net/http"
"net/url"
"sync"
"time"
)
const (
ProtocolVersionHybi13 = 13
ProtocolVersionHybi = ProtocolVersionHybi13
SupportedProtocolVersion = "13"
ContinuationFrame = 0
TextFrame = 1
BinaryFrame = 2
CloseFrame = 8
PingFrame = 9
PongFrame = 10
UnknownFrame = 255
)
// ProtocolError represents WebSocket protocol errors.
type ProtocolError struct {
ErrorString string
}
func (err *ProtocolError) Error() string { return err.ErrorString }
var (
ErrBadProtocolVersion = &ProtocolError{"bad protocol version"}
ErrBadScheme = &ProtocolError{"bad scheme"}
ErrBadStatus = &ProtocolError{"bad status"}
ErrBadUpgrade = &ProtocolError{"missing or bad upgrade"}
ErrBadWebSocketOrigin = &ProtocolError{"missing or bad WebSocket-Origin"}
ErrBadWebSocketLocation = &ProtocolError{"missing or bad WebSocket-Location"}
ErrBadWebSocketProtocol = &ProtocolError{"missing or bad WebSocket-Protocol"}
ErrBadWebSocketVersion = &ProtocolError{"missing or bad WebSocket Version"}
ErrChallengeResponse = &ProtocolError{"mismatch challenge/response"}
ErrBadFrame = &ProtocolError{"bad frame"}
ErrBadFrameBoundary = &ProtocolError{"not on frame boundary"}
ErrNotWebSocket = &ProtocolError{"not websocket protocol"}
ErrBadRequestMethod = &ProtocolError{"bad method"}
ErrNotSupported = &ProtocolError{"not supported"}
)
// Addr is an implementation of net.Addr for WebSocket.
type Addr struct {
*url.URL
}
// Network returns the network type for a WebSocket, "websocket".
func (addr *Addr) Network() string { return "websocket" }
// Config is a WebSocket configuration
type Config struct {
// A WebSocket server address.
Location *url.URL
// A Websocket client origin.
Origin *url.URL
// WebSocket subprotocols.
Protocol []string
// WebSocket protocol version.
Version int
// TLS config for secure WebSocket (wss).
TlsConfig *tls.Config
// Additional header fields to be sent in WebSocket opening handshake.
Header http.Header
handshakeData map[string]string
}
// serverHandshaker is an interface to handle WebSocket server side handshake.
type serverHandshaker interface {
// ReadHandshake reads handshake request message from client.
// Returns http response code and error if any.
ReadHandshake(buf *bufio.Reader, req *http.Request) (code int, err error)
// AcceptHandshake accepts the client handshake request and sends
// handshake response back to client.
AcceptHandshake(buf *bufio.Writer) (err error)
// NewServerConn creates a new WebSocket connection.
NewServerConn(buf *bufio.ReadWriter, rwc io.ReadWriteCloser, request *http.Request) (conn *Conn)
}
// frameReader is an interface to read a WebSocket frame.
type frameReader interface {
// Reader is to read payload of the frame.
io.Reader
// PayloadType returns payload type.
PayloadType() byte
// HeaderReader returns a reader to read header of the frame.
HeaderReader() io.Reader
// TrailerReader returns a reader to read trailer of the frame.
// If it returns nil, there is no trailer in the frame.
TrailerReader() io.Reader
// Len returns total length of the frame, including header and trailer.
Len() int
}
// frameReaderFactory is an interface to creates new frame reader.
type frameReaderFactory interface {
NewFrameReader() (r frameReader, err error)
}
// frameWriter is an interface to write a WebSocket frame.
type frameWriter interface {
// Writer is to write payload of the frame.
io.WriteCloser
}
// frameWriterFactory is an interface to create new frame writer.
type frameWriterFactory interface {
NewFrameWriter(payloadType byte) (w frameWriter, err error)
}
type frameHandler interface {
HandleFrame(frame frameReader) (r frameReader, err error)
WriteClose(status int) (err error)
}
// Conn represents a WebSocket connection.
type Conn struct {
config *Config
request *http.Request
buf *bufio.ReadWriter
rwc io.ReadWriteCloser
rio sync.Mutex
frameReaderFactory
frameReader
wio sync.Mutex
frameWriterFactory
frameHandler
PayloadType byte
defaultCloseStatus int
}
// Read implements the io.Reader interface:
// it reads data of a frame from the WebSocket connection.
// if msg is not large enough for the frame data, it fills the msg and next Read
// will read the rest of the frame data.
// it reads Text frame or Binary frame.
func (ws *Conn) Read(msg []byte) (n int, err error) {
ws.rio.Lock()
defer ws.rio.Unlock()
again:
if ws.frameReader == nil {
frame, err := ws.frameReaderFactory.NewFrameReader()
if err != nil {
return 0, err
}
ws.frameReader, err = ws.frameHandler.HandleFrame(frame)
if err != nil {
return 0, err
}
if ws.frameReader == nil {
goto again
}
}
n, err = ws.frameReader.Read(msg)
if err == io.EOF {
if trailer := ws.frameReader.TrailerReader(); trailer != nil {
io.Copy(ioutil.Discard, trailer)
}
ws.frameReader = nil
goto again
}
return n, err
}
// Write implements the io.Writer interface:
// it writes data as a frame to the WebSocket connection.
func (ws *Conn) Write(msg []byte) (n int, err error) {
ws.wio.Lock()
defer ws.wio.Unlock()
w, err := ws.frameWriterFactory.NewFrameWriter(ws.PayloadType)
if err != nil {
return 0, err
}
n, err = w.Write(msg)
w.Close()
if err != nil {
return n, err
}
return n, err
}
// Close implements the io.Closer interface.
func (ws *Conn) Close() error {
err := ws.frameHandler.WriteClose(ws.defaultCloseStatus)
if err != nil {
return err
}
return ws.rwc.Close()
}
func (ws *Conn) IsClientConn() bool { return ws.request == nil }
func (ws *Conn) IsServerConn() bool { return ws.request != nil }
// LocalAddr returns the WebSocket Origin for the connection for client, or
// the WebSocket location for server.
func (ws *Conn) LocalAddr() net.Addr {
if ws.IsClientConn() {
return &Addr{ws.config.Origin}
}
return &Addr{ws.config.Location}
}
// RemoteAddr returns the WebSocket location for the connection for client, or
// the Websocket Origin for server.
func (ws *Conn) RemoteAddr() net.Addr {
if ws.IsClientConn() {
return &Addr{ws.config.Location}
}
return &Addr{ws.config.Origin}
}
var errSetDeadline = errors.New("websocket: cannot set deadline: not using a net.Conn")
// SetDeadline sets the connection's network read & write deadlines.
func (ws *Conn) SetDeadline(t time.Time) error {
if conn, ok := ws.rwc.(net.Conn); ok {
return conn.SetDeadline(t)
}
return errSetDeadline
}
// SetReadDeadline sets the connection's network read deadline.
func (ws *Conn) SetReadDeadline(t time.Time) error {
if conn, ok := ws.rwc.(net.Conn); ok {
return conn.SetReadDeadline(t)
}
return errSetDeadline
}
// SetWriteDeadline sets the connection's network write deadline.
func (ws *Conn) SetWriteDeadline(t time.Time) error {
if conn, ok := ws.rwc.(net.Conn); ok {
return conn.SetWriteDeadline(t)
}
return errSetDeadline
}
// Config returns the WebSocket config.
func (ws *Conn) Config() *Config { return ws.config }
// Request returns the http request upgraded to the WebSocket.
// It is nil for client side.
func (ws *Conn) Request() *http.Request { return ws.request }
// Codec represents a symmetric pair of functions that implement a codec.
type Codec struct {
Marshal func(v interface{}) (data []byte, payloadType byte, err error)
Unmarshal func(data []byte, payloadType byte, v interface{}) (err error)
}
// Send sends v marshaled by cd.Marshal as single frame to ws.
func (cd Codec) Send(ws *Conn, v interface{}) (err error) {
data, payloadType, err := cd.Marshal(v)
if err != nil {
return err
}
ws.wio.Lock()
defer ws.wio.Unlock()
w, err := ws.frameWriterFactory.NewFrameWriter(payloadType)
if err != nil {
return err
}
_, err = w.Write(data)
w.Close()
return err
}
// Receive receives single frame from ws, unmarshaled by cd.Unmarshal and stores in v.
func (cd Codec) Receive(ws *Conn, v interface{}) (err error) {
ws.rio.Lock()
defer ws.rio.Unlock()
if ws.frameReader != nil {
_, err = io.Copy(ioutil.Discard, ws.frameReader)
if err != nil {
return err
}
ws.frameReader = nil
}
again:
frame, err := ws.frameReaderFactory.NewFrameReader()
if err != nil {
return err
}
frame, err = ws.frameHandler.HandleFrame(frame)
if err != nil {
return err
}
if frame == nil {
goto again
}
payloadType := frame.PayloadType()
data, err := ioutil.ReadAll(frame)
if err != nil {
return err
}
return cd.Unmarshal(data, payloadType, v)
}
func marshal(v interface{}) (msg []byte, payloadType byte, err error) {
switch data := v.(type) {
case string:
return []byte(data), TextFrame, nil
case []byte:
return data, BinaryFrame, nil
}
return nil, UnknownFrame, ErrNotSupported
}
func unmarshal(msg []byte, payloadType byte, v interface{}) (err error) {
switch data := v.(type) {
case *string:
*data = string(msg)
return nil
case *[]byte:
*data = msg
return nil
}
return ErrNotSupported
}
/*
Message is a codec to send/receive text/binary data in a frame on WebSocket connection.
To send/receive text frame, use string type.
To send/receive binary frame, use []byte type.
Trivial usage:
import "websocket"
// receive text frame
var message string
websocket.Message.Receive(ws, &message)
// send text frame
message = "hello"
websocket.Message.Send(ws, message)
// receive binary frame
var data []byte
websocket.Message.Receive(ws, &data)
// send binary frame
data = []byte{0, 1, 2}
websocket.Message.Send(ws, data)
*/
var Message = Codec{marshal, unmarshal}
func jsonMarshal(v interface{}) (msg []byte, payloadType byte, err error) {
msg, err = json.Marshal(v)
return msg, TextFrame, err
}
func jsonUnmarshal(msg []byte, payloadType byte, v interface{}) (err error) {
return json.Unmarshal(msg, v)
}
/*
JSON is a codec to send/receive JSON data in a frame from a WebSocket connection.
Trivial usage:
import "websocket"
type T struct {
Msg string
Count int
}
// receive JSON type T
var data T
websocket.JSON.Receive(ws, &data)
// send JSON type T
websocket.JSON.Send(ws, data)
*/
var JSON = Codec{jsonMarshal, jsonUnmarshal}

341
third_party/src/code.google.com/p/go.net/websocket/websocket_test.go vendored Normal file
View File

@ -0,0 +1,341 @@
// Copyright 2009 The Go Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
package websocket
import (
"bytes"
"fmt"
"io"
"log"
"net"
"net/http"
"net/http/httptest"
"net/url"
"strings"
"sync"
"testing"
)
var serverAddr string
var once sync.Once
func echoServer(ws *Conn) { io.Copy(ws, ws) }
type Count struct {
S string
N int
}
func countServer(ws *Conn) {
for {
var count Count
err := JSON.Receive(ws, &count)
if err != nil {
return
}
count.N++
count.S = strings.Repeat(count.S, count.N)
err = JSON.Send(ws, count)
if err != nil {
return
}
}
}
func subProtocolHandshake(config *Config, req *http.Request) error {
for _, proto := range config.Protocol {
if proto == "chat" {
config.Protocol = []string{proto}
return nil
}
}
return ErrBadWebSocketProtocol
}
func subProtoServer(ws *Conn) {
for _, proto := range ws.Config().Protocol {
io.WriteString(ws, proto)
}
}
func startServer() {
http.Handle("/echo", Handler(echoServer))
http.Handle("/count", Handler(countServer))
subproto := Server{
Handshake: subProtocolHandshake,
Handler: Handler(subProtoServer),
}
http.Handle("/subproto", subproto)
server := httptest.NewServer(nil)
serverAddr = server.Listener.Addr().String()
log.Print("Test WebSocket server listening on ", serverAddr)
}
func newConfig(t *testing.T, path string) *Config {
config, _ := NewConfig(fmt.Sprintf("ws://%s%s", serverAddr, path), "http://localhost")
return config
}
func TestEcho(t *testing.T) {
once.Do(startServer)
// websocket.Dial()
client, err := net.Dial("tcp", serverAddr)
if err != nil {
t.Fatal("dialing", err)
}
conn, err := NewClient(newConfig(t, "/echo"), client)
if err != nil {
t.Errorf("WebSocket handshake error: %v", err)
return
}
msg := []byte("hello, world\n")
if _, err := conn.Write(msg); err != nil {
t.Errorf("Write: %v", err)
}
var actual_msg = make([]byte, 512)
n, err := conn.Read(actual_msg)
if err != nil {
t.Errorf("Read: %v", err)
}
actual_msg = actual_msg[0:n]
if !bytes.Equal(msg, actual_msg) {
t.Errorf("Echo: expected %q got %q", msg, actual_msg)
}
conn.Close()
}
func TestAddr(t *testing.T) {
once.Do(startServer)
// websocket.Dial()
client, err := net.Dial("tcp", serverAddr)
if err != nil {
t.Fatal("dialing", err)
}
conn, err := NewClient(newConfig(t, "/echo"), client)
if err != nil {
t.Errorf("WebSocket handshake error: %v", err)
return
}
ra := conn.RemoteAddr().String()
if !strings.HasPrefix(ra, "ws://") || !strings.HasSuffix(ra, "/echo") {
t.Errorf("Bad remote addr: %v", ra)
}
la := conn.LocalAddr().String()
if !strings.HasPrefix(la, "http://") {
t.Errorf("Bad local addr: %v", la)
}
conn.Close()
}
func TestCount(t *testing.T) {
once.Do(startServer)
// websocket.Dial()
client, err := net.Dial("tcp", serverAddr)
if err != nil {
t.Fatal("dialing", err)
}
conn, err := NewClient(newConfig(t, "/count"), client)
if err != nil {
t.Errorf("WebSocket handshake error: %v", err)
return
}
var count Count
count.S = "hello"
if err := JSON.Send(conn, count); err != nil {
t.Errorf("Write: %v", err)
}
if err := JSON.Receive(conn, &count); err != nil {
t.Errorf("Read: %v", err)
}
if count.N != 1 {
t.Errorf("count: expected %d got %d", 1, count.N)
}
if count.S != "hello" {
t.Errorf("count: expected %q got %q", "hello", count.S)
}
if err := JSON.Send(conn, count); err != nil {
t.Errorf("Write: %v", err)
}
if err := JSON.Receive(conn, &count); err != nil {
t.Errorf("Read: %v", err)
}
if count.N != 2 {
t.Errorf("count: expected %d got %d", 2, count.N)
}
if count.S != "hellohello" {
t.Errorf("count: expected %q got %q", "hellohello", count.S)
}
conn.Close()
}
func TestWithQuery(t *testing.T) {
once.Do(startServer)
client, err := net.Dial("tcp", serverAddr)
if err != nil {
t.Fatal("dialing", err)
}
config := newConfig(t, "/echo")
config.Location, err = url.ParseRequestURI(fmt.Sprintf("ws://%s/echo?q=v", serverAddr))
if err != nil {
t.Fatal("location url", err)
}
ws, err := NewClient(config, client)
if err != nil {
t.Errorf("WebSocket handshake: %v", err)
return
}
ws.Close()
}
func testWithProtocol(t *testing.T, subproto []string) (string, error) {
once.Do(startServer)
client, err := net.Dial("tcp", serverAddr)
if err != nil {
t.Fatal("dialing", err)
}
config := newConfig(t, "/subproto")
config.Protocol = subproto
ws, err := NewClient(config, client)
if err != nil {
return "", err
}
msg := make([]byte, 16)
n, err := ws.Read(msg)
if err != nil {
return "", err
}
ws.Close()
return string(msg[:n]), nil
}
func TestWithProtocol(t *testing.T) {
proto, err := testWithProtocol(t, []string{"chat"})
if err != nil {
t.Errorf("SubProto: unexpected error: %v", err)
}
if proto != "chat" {
t.Errorf("SubProto: expected %q, got %q", "chat", proto)
}
}
func TestWithTwoProtocol(t *testing.T) {
proto, err := testWithProtocol(t, []string{"test", "chat"})
if err != nil {
t.Errorf("SubProto: unexpected error: %v", err)
}
if proto != "chat" {
t.Errorf("SubProto: expected %q, got %q", "chat", proto)
}
}
func TestWithBadProtocol(t *testing.T) {
_, err := testWithProtocol(t, []string{"test"})
if err != ErrBadStatus {
t.Errorf("SubProto: expected %v, got %v", ErrBadStatus, err)
}
}
func TestHTTP(t *testing.T) {
once.Do(startServer)
// If the client did not send a handshake that matches the protocol
// specification, the server MUST return an HTTP response with an
// appropriate error code (such as 400 Bad Request)
resp, err := http.Get(fmt.Sprintf("http://%s/echo", serverAddr))
if err != nil {
t.Errorf("Get: error %#v", err)
return
}
if resp == nil {
t.Error("Get: resp is null")
return
}
if resp.StatusCode != http.StatusBadRequest {
t.Errorf("Get: expected %q got %q", http.StatusBadRequest, resp.StatusCode)
}
}
func TestTrailingSpaces(t *testing.T) {
// http://code.google.com/p/go/issues/detail?id=955
// The last runs of this create keys with trailing spaces that should not be
// generated by the client.
once.Do(startServer)
config := newConfig(t, "/echo")
for i := 0; i < 30; i++ {
// body
ws, err := DialConfig(config)
if err != nil {
t.Errorf("Dial #%d failed: %v", i, err)
break
}
ws.Close()
}
}
func TestDialConfigBadVersion(t *testing.T) {
once.Do(startServer)
config := newConfig(t, "/echo")
config.Version = 1234
_, err := DialConfig(config)
if dialerr, ok := err.(*DialError); ok {
if dialerr.Err != ErrBadProtocolVersion {
t.Errorf("dial expected err %q but got %q", ErrBadProtocolVersion, dialerr.Err)
}
}
}
func TestSmallBuffer(t *testing.T) {
// http://code.google.com/p/go/issues/detail?id=1145
// Read should be able to handle reading a fragment of a frame.
once.Do(startServer)
// websocket.Dial()
client, err := net.Dial("tcp", serverAddr)
if err != nil {
t.Fatal("dialing", err)
}
conn, err := NewClient(newConfig(t, "/echo"), client)
if err != nil {
t.Errorf("WebSocket handshake error: %v", err)
return
}
msg := []byte("hello, world\n")
if _, err := conn.Write(msg); err != nil {
t.Errorf("Write: %v", err)
}
var small_msg = make([]byte, 8)
n, err := conn.Read(small_msg)
if err != nil {
t.Errorf("Read: %v", err)
}
if !bytes.Equal(msg[:len(small_msg)], small_msg) {
t.Errorf("Echo: expected %q got %q", msg[:len(small_msg)], small_msg)
}
var second_msg = make([]byte, len(msg))
n, err = conn.Read(second_msg)
if err != nil {
t.Errorf("Read: %v", err)
}
second_msg = second_msg[0:n]
if !bytes.Equal(msg[len(small_msg):], second_msg) {
t.Errorf("Echo: expected %q got %q", msg[len(small_msg):], second_msg)
}
conn.Close()
}