conn.go 24 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951
  1. // Copyright 2013 The Gorilla WebSocket Authors. All rights reserved.
  2. // Use of this source code is governed by a BSD-style
  3. // license that can be found in the LICENSE file.
  4. package websocket
  5. import (
  6. "bufio"
  7. "encoding/binary"
  8. "errors"
  9. "io"
  10. "io/ioutil"
  11. "math/rand"
  12. "net"
  13. "strconv"
  14. "time"
  15. "unicode/utf8"
  16. )
  17. const (
  18. maxFrameHeaderSize = 2 + 8 + 4 // Fixed header + length + mask
  19. maxControlFramePayloadSize = 125
  20. finalBit = 1 << 7
  21. maskBit = 1 << 7
  22. writeWait = time.Second
  23. defaultReadBufferSize = 4096
  24. defaultWriteBufferSize = 4096
  25. continuationFrame = 0
  26. noFrame = -1
  27. )
  28. // Close codes defined in RFC 6455, section 11.7.
  29. const (
  30. CloseNormalClosure = 1000
  31. CloseGoingAway = 1001
  32. CloseProtocolError = 1002
  33. CloseUnsupportedData = 1003
  34. CloseNoStatusReceived = 1005
  35. CloseAbnormalClosure = 1006
  36. CloseInvalidFramePayloadData = 1007
  37. ClosePolicyViolation = 1008
  38. CloseMessageTooBig = 1009
  39. CloseMandatoryExtension = 1010
  40. CloseInternalServerErr = 1011
  41. CloseServiceRestart = 1012
  42. CloseTryAgainLater = 1013
  43. CloseTLSHandshake = 1015
  44. )
  45. // The message types are defined in RFC 6455, section 11.8.
  46. const (
  47. // TextMessage denotes a text data message. The text message payload is
  48. // interpreted as UTF-8 encoded text data.
  49. TextMessage = 1
  50. // BinaryMessage denotes a binary data message.
  51. BinaryMessage = 2
  52. // CloseMessage denotes a close control message. The optional message
  53. // payload contains a numeric code and text. Use the FormatCloseMessage
  54. // function to format a close message payload.
  55. CloseMessage = 8
  56. // PingMessage denotes a ping control message. The optional message payload
  57. // is UTF-8 encoded text.
  58. PingMessage = 9
  59. // PongMessage denotes a ping control message. The optional message payload
  60. // is UTF-8 encoded text.
  61. PongMessage = 10
  62. )
  63. // ErrCloseSent is returned when the application writes a message to the
  64. // connection after sending a close message.
  65. var ErrCloseSent = errors.New("websocket: close sent")
  66. // ErrReadLimit is returned when reading a message that is larger than the
  67. // read limit set for the connection.
  68. var ErrReadLimit = errors.New("websocket: read limit exceeded")
  69. // netError satisfies the net Error interface.
  70. type netError struct {
  71. msg string
  72. temporary bool
  73. timeout bool
  74. }
  75. func (e *netError) Error() string { return e.msg }
  76. func (e *netError) Temporary() bool { return e.temporary }
  77. func (e *netError) Timeout() bool { return e.timeout }
  78. // CloseError represents close frame.
  79. type CloseError struct {
  80. // Code is defined in RFC 6455, section 11.7.
  81. Code int
  82. // Text is the optional text payload.
  83. Text string
  84. }
  85. func (e *CloseError) Error() string {
  86. s := []byte("websocket: close ")
  87. s = strconv.AppendInt(s, int64(e.Code), 10)
  88. switch e.Code {
  89. case CloseNormalClosure:
  90. s = append(s, " (normal)"...)
  91. case CloseGoingAway:
  92. s = append(s, " (going away)"...)
  93. case CloseProtocolError:
  94. s = append(s, " (protocol error)"...)
  95. case CloseUnsupportedData:
  96. s = append(s, " (unsupported data)"...)
  97. case CloseNoStatusReceived:
  98. s = append(s, " (no status)"...)
  99. case CloseAbnormalClosure:
  100. s = append(s, " (abnormal closure)"...)
  101. case CloseInvalidFramePayloadData:
  102. s = append(s, " (invalid payload data)"...)
  103. case ClosePolicyViolation:
  104. s = append(s, " (policy violation)"...)
  105. case CloseMessageTooBig:
  106. s = append(s, " (message too big)"...)
  107. case CloseMandatoryExtension:
  108. s = append(s, " (mandatory extension missing)"...)
  109. case CloseInternalServerErr:
  110. s = append(s, " (internal server error)"...)
  111. case CloseTLSHandshake:
  112. s = append(s, " (TLS handshake error)"...)
  113. }
  114. if e.Text != "" {
  115. s = append(s, ": "...)
  116. s = append(s, e.Text...)
  117. }
  118. return string(s)
  119. }
  120. // IsCloseError returns boolean indicating whether the error is a *CloseError
  121. // with one of the specified codes.
  122. func IsCloseError(err error, codes ...int) bool {
  123. if e, ok := err.(*CloseError); ok {
  124. for _, code := range codes {
  125. if e.Code == code {
  126. return true
  127. }
  128. }
  129. }
  130. return false
  131. }
  132. // IsUnexpectedCloseError returns boolean indicating whether the error is a
  133. // *CloseError with a code not in the list of expected codes.
  134. func IsUnexpectedCloseError(err error, expectedCodes ...int) bool {
  135. if e, ok := err.(*CloseError); ok {
  136. for _, code := range expectedCodes {
  137. if e.Code == code {
  138. return false
  139. }
  140. }
  141. return true
  142. }
  143. return false
  144. }
  145. var (
  146. errWriteTimeout = &netError{msg: "websocket: write timeout", timeout: true, temporary: true}
  147. errUnexpectedEOF = &CloseError{Code: CloseAbnormalClosure, Text: io.ErrUnexpectedEOF.Error()}
  148. errBadWriteOpCode = errors.New("websocket: bad write message type")
  149. errWriteClosed = errors.New("websocket: write closed")
  150. errInvalidControlFrame = errors.New("websocket: invalid control frame")
  151. )
  152. func hideTempErr(err error) error {
  153. if e, ok := err.(net.Error); ok && e.Temporary() {
  154. err = &netError{msg: e.Error(), timeout: e.Timeout()}
  155. }
  156. return err
  157. }
  158. func isControl(frameType int) bool {
  159. return frameType == CloseMessage || frameType == PingMessage || frameType == PongMessage
  160. }
  161. func isData(frameType int) bool {
  162. return frameType == TextMessage || frameType == BinaryMessage
  163. }
  164. var validReceivedCloseCodes = map[int]bool{
  165. // see http://www.iana.org/assignments/websocket/websocket.xhtml#close-code-number
  166. CloseNormalClosure: true,
  167. CloseGoingAway: true,
  168. CloseProtocolError: true,
  169. CloseUnsupportedData: true,
  170. CloseNoStatusReceived: false,
  171. CloseAbnormalClosure: false,
  172. CloseInvalidFramePayloadData: true,
  173. ClosePolicyViolation: true,
  174. CloseMessageTooBig: true,
  175. CloseMandatoryExtension: true,
  176. CloseInternalServerErr: true,
  177. CloseServiceRestart: true,
  178. CloseTryAgainLater: true,
  179. CloseTLSHandshake: false,
  180. }
  181. func isValidReceivedCloseCode(code int) bool {
  182. return validReceivedCloseCodes[code] || (code >= 3000 && code <= 4999)
  183. }
  184. func maskBytes(key [4]byte, pos int, b []byte) int {
  185. for i := range b {
  186. b[i] ^= key[pos&3]
  187. pos++
  188. }
  189. return pos & 3
  190. }
  191. func newMaskKey() [4]byte {
  192. n := rand.Uint32()
  193. return [4]byte{byte(n), byte(n >> 8), byte(n >> 16), byte(n >> 24)}
  194. }
  195. // Conn represents a WebSocket connection.
  196. type Conn struct {
  197. conn net.Conn
  198. isServer bool
  199. subprotocol string
  200. // Write fields
  201. mu chan bool // used as mutex to protect write to conn and closeSent
  202. closeSent bool // true if close message was sent
  203. // Message writer fields.
  204. writeErr error
  205. writeBuf []byte // frame is constructed in this buffer.
  206. writePos int // end of data in writeBuf.
  207. writeFrameType int // type of the current frame.
  208. writeDeadline time.Time
  209. isWriting bool // for best-effort concurrent write detection
  210. messageWriter *messageWriter // the current writer
  211. // Read fields
  212. readErr error
  213. br *bufio.Reader
  214. readRemaining int64 // bytes remaining in current frame.
  215. readFinal bool // true the current message has more frames.
  216. readLength int64 // Message size.
  217. readLimit int64 // Maximum message size.
  218. readMaskPos int
  219. readMaskKey [4]byte
  220. handlePong func(string) error
  221. handlePing func(string) error
  222. readErrCount int
  223. messageReader *messageReader // the current reader
  224. }
  225. func newConn(conn net.Conn, isServer bool, readBufferSize, writeBufferSize int) *Conn {
  226. mu := make(chan bool, 1)
  227. mu <- true
  228. if readBufferSize == 0 {
  229. readBufferSize = defaultReadBufferSize
  230. }
  231. if readBufferSize < maxControlFramePayloadSize {
  232. readBufferSize = maxControlFramePayloadSize
  233. }
  234. if writeBufferSize == 0 {
  235. writeBufferSize = defaultWriteBufferSize
  236. }
  237. c := &Conn{
  238. isServer: isServer,
  239. br: bufio.NewReaderSize(conn, readBufferSize),
  240. conn: conn,
  241. mu: mu,
  242. readFinal: true,
  243. writeBuf: make([]byte, writeBufferSize+maxFrameHeaderSize),
  244. writeFrameType: noFrame,
  245. writePos: maxFrameHeaderSize,
  246. }
  247. c.SetPingHandler(nil)
  248. c.SetPongHandler(nil)
  249. return c
  250. }
  251. // Subprotocol returns the negotiated protocol for the connection.
  252. func (c *Conn) Subprotocol() string {
  253. return c.subprotocol
  254. }
  255. // Close closes the underlying network connection without sending or waiting for a close frame.
  256. func (c *Conn) Close() error {
  257. return c.conn.Close()
  258. }
  259. // LocalAddr returns the local network address.
  260. func (c *Conn) LocalAddr() net.Addr {
  261. return c.conn.LocalAddr()
  262. }
  263. // RemoteAddr returns the remote network address.
  264. func (c *Conn) RemoteAddr() net.Addr {
  265. return c.conn.RemoteAddr()
  266. }
  267. // Write methods
  268. func (c *Conn) write(frameType int, deadline time.Time, bufs ...[]byte) error {
  269. <-c.mu
  270. defer func() { c.mu <- true }()
  271. if c.closeSent {
  272. return ErrCloseSent
  273. } else if frameType == CloseMessage {
  274. c.closeSent = true
  275. }
  276. c.conn.SetWriteDeadline(deadline)
  277. for _, buf := range bufs {
  278. if len(buf) > 0 {
  279. n, err := c.conn.Write(buf)
  280. if n != len(buf) {
  281. // Close on partial write.
  282. c.conn.Close()
  283. }
  284. if err != nil {
  285. return err
  286. }
  287. }
  288. }
  289. return nil
  290. }
  291. // WriteControl writes a control message with the given deadline. The allowed
  292. // message types are CloseMessage, PingMessage and PongMessage.
  293. func (c *Conn) WriteControl(messageType int, data []byte, deadline time.Time) error {
  294. if !isControl(messageType) {
  295. return errBadWriteOpCode
  296. }
  297. if len(data) > maxControlFramePayloadSize {
  298. return errInvalidControlFrame
  299. }
  300. b0 := byte(messageType) | finalBit
  301. b1 := byte(len(data))
  302. if !c.isServer {
  303. b1 |= maskBit
  304. }
  305. buf := make([]byte, 0, maxFrameHeaderSize+maxControlFramePayloadSize)
  306. buf = append(buf, b0, b1)
  307. if c.isServer {
  308. buf = append(buf, data...)
  309. } else {
  310. key := newMaskKey()
  311. buf = append(buf, key[:]...)
  312. buf = append(buf, data...)
  313. maskBytes(key, 0, buf[6:])
  314. }
  315. d := time.Hour * 1000
  316. if !deadline.IsZero() {
  317. d = deadline.Sub(time.Now())
  318. if d < 0 {
  319. return errWriteTimeout
  320. }
  321. }
  322. timer := time.NewTimer(d)
  323. select {
  324. case <-c.mu:
  325. timer.Stop()
  326. case <-timer.C:
  327. return errWriteTimeout
  328. }
  329. defer func() { c.mu <- true }()
  330. if c.closeSent {
  331. return ErrCloseSent
  332. } else if messageType == CloseMessage {
  333. c.closeSent = true
  334. }
  335. c.conn.SetWriteDeadline(deadline)
  336. n, err := c.conn.Write(buf)
  337. if n != 0 && n != len(buf) {
  338. c.conn.Close()
  339. }
  340. return hideTempErr(err)
  341. }
  342. // NextWriter returns a writer for the next message to send. The writer's Close
  343. // method flushes the complete message to the network.
  344. //
  345. // There can be at most one open writer on a connection. NextWriter closes the
  346. // previous writer if the application has not already done so.
  347. func (c *Conn) NextWriter(messageType int) (io.WriteCloser, error) {
  348. if c.writeErr != nil {
  349. return nil, c.writeErr
  350. }
  351. if c.writeFrameType != noFrame {
  352. if err := c.flushFrame(true, nil); err != nil {
  353. return nil, err
  354. }
  355. }
  356. if !isControl(messageType) && !isData(messageType) {
  357. return nil, errBadWriteOpCode
  358. }
  359. c.writeFrameType = messageType
  360. w := &messageWriter{c}
  361. c.messageWriter = w
  362. return w, nil
  363. }
  364. func (c *Conn) flushFrame(final bool, extra []byte) error {
  365. length := c.writePos - maxFrameHeaderSize + len(extra)
  366. // Check for invalid control frames.
  367. if isControl(c.writeFrameType) &&
  368. (!final || length > maxControlFramePayloadSize) {
  369. c.messageWriter = nil
  370. c.writeFrameType = noFrame
  371. c.writePos = maxFrameHeaderSize
  372. return errInvalidControlFrame
  373. }
  374. b0 := byte(c.writeFrameType)
  375. if final {
  376. b0 |= finalBit
  377. }
  378. b1 := byte(0)
  379. if !c.isServer {
  380. b1 |= maskBit
  381. }
  382. // Assume that the frame starts at beginning of c.writeBuf.
  383. framePos := 0
  384. if c.isServer {
  385. // Adjust up if mask not included in the header.
  386. framePos = 4
  387. }
  388. switch {
  389. case length >= 65536:
  390. c.writeBuf[framePos] = b0
  391. c.writeBuf[framePos+1] = b1 | 127
  392. binary.BigEndian.PutUint64(c.writeBuf[framePos+2:], uint64(length))
  393. case length > 125:
  394. framePos += 6
  395. c.writeBuf[framePos] = b0
  396. c.writeBuf[framePos+1] = b1 | 126
  397. binary.BigEndian.PutUint16(c.writeBuf[framePos+2:], uint16(length))
  398. default:
  399. framePos += 8
  400. c.writeBuf[framePos] = b0
  401. c.writeBuf[framePos+1] = b1 | byte(length)
  402. }
  403. if !c.isServer {
  404. key := newMaskKey()
  405. copy(c.writeBuf[maxFrameHeaderSize-4:], key[:])
  406. maskBytes(key, 0, c.writeBuf[maxFrameHeaderSize:c.writePos])
  407. if len(extra) > 0 {
  408. c.writeErr = errors.New("websocket: internal error, extra used in client mode")
  409. return c.writeErr
  410. }
  411. }
  412. // Write the buffers to the connection with best-effort detection of
  413. // concurrent writes. See the concurrency section in the package
  414. // documentation for more info.
  415. if c.isWriting {
  416. panic("concurrent write to websocket connection")
  417. }
  418. c.isWriting = true
  419. c.writeErr = c.write(c.writeFrameType, c.writeDeadline, c.writeBuf[framePos:c.writePos], extra)
  420. if !c.isWriting {
  421. panic("concurrent write to websocket connection")
  422. }
  423. c.isWriting = false
  424. // Setup for next frame.
  425. c.writePos = maxFrameHeaderSize
  426. c.writeFrameType = continuationFrame
  427. if final {
  428. c.messageWriter = nil
  429. c.writeFrameType = noFrame
  430. }
  431. return c.writeErr
  432. }
  433. type messageWriter struct{ c *Conn }
  434. func (w *messageWriter) err() error {
  435. c := w.c
  436. if c.messageWriter != w {
  437. return errWriteClosed
  438. }
  439. if c.writeErr != nil {
  440. return c.writeErr
  441. }
  442. return nil
  443. }
  444. func (w *messageWriter) ncopy(max int) (int, error) {
  445. n := len(w.c.writeBuf) - w.c.writePos
  446. if n <= 0 {
  447. if err := w.c.flushFrame(false, nil); err != nil {
  448. return 0, err
  449. }
  450. n = len(w.c.writeBuf) - w.c.writePos
  451. }
  452. if n > max {
  453. n = max
  454. }
  455. return n, nil
  456. }
  457. func (w *messageWriter) write(final bool, p []byte) (int, error) {
  458. if err := w.err(); err != nil {
  459. return 0, err
  460. }
  461. if len(p) > 2*len(w.c.writeBuf) && w.c.isServer {
  462. // Don't buffer large messages.
  463. err := w.c.flushFrame(final, p)
  464. if err != nil {
  465. return 0, err
  466. }
  467. return len(p), nil
  468. }
  469. nn := len(p)
  470. for len(p) > 0 {
  471. n, err := w.ncopy(len(p))
  472. if err != nil {
  473. return 0, err
  474. }
  475. copy(w.c.writeBuf[w.c.writePos:], p[:n])
  476. w.c.writePos += n
  477. p = p[n:]
  478. }
  479. return nn, nil
  480. }
  481. func (w *messageWriter) Write(p []byte) (int, error) {
  482. return w.write(false, p)
  483. }
  484. func (w *messageWriter) WriteString(p string) (int, error) {
  485. if err := w.err(); err != nil {
  486. return 0, err
  487. }
  488. nn := len(p)
  489. for len(p) > 0 {
  490. n, err := w.ncopy(len(p))
  491. if err != nil {
  492. return 0, err
  493. }
  494. copy(w.c.writeBuf[w.c.writePos:], p[:n])
  495. w.c.writePos += n
  496. p = p[n:]
  497. }
  498. return nn, nil
  499. }
  500. func (w *messageWriter) ReadFrom(r io.Reader) (nn int64, err error) {
  501. if err := w.err(); err != nil {
  502. return 0, err
  503. }
  504. for {
  505. if w.c.writePos == len(w.c.writeBuf) {
  506. err = w.c.flushFrame(false, nil)
  507. if err != nil {
  508. break
  509. }
  510. }
  511. var n int
  512. n, err = r.Read(w.c.writeBuf[w.c.writePos:])
  513. w.c.writePos += n
  514. nn += int64(n)
  515. if err != nil {
  516. if err == io.EOF {
  517. err = nil
  518. }
  519. break
  520. }
  521. }
  522. return nn, err
  523. }
  524. func (w *messageWriter) Close() error {
  525. if err := w.err(); err != nil {
  526. return err
  527. }
  528. return w.c.flushFrame(true, nil)
  529. }
  530. // WriteMessage is a helper method for getting a writer using NextWriter,
  531. // writing the message and closing the writer.
  532. func (c *Conn) WriteMessage(messageType int, data []byte) error {
  533. w, err := c.NextWriter(messageType)
  534. if err != nil {
  535. return err
  536. }
  537. if _, ok := w.(*messageWriter); ok && c.isServer {
  538. // Optimize write as a single frame.
  539. n := copy(c.writeBuf[c.writePos:], data)
  540. c.writePos += n
  541. data = data[n:]
  542. err = c.flushFrame(true, data)
  543. return err
  544. }
  545. if _, err = w.Write(data); err != nil {
  546. return err
  547. }
  548. return w.Close()
  549. }
  550. // SetWriteDeadline sets the write deadline on the underlying network
  551. // connection. After a write has timed out, the websocket state is corrupt and
  552. // all future writes will return an error. A zero value for t means writes will
  553. // not time out.
  554. func (c *Conn) SetWriteDeadline(t time.Time) error {
  555. c.writeDeadline = t
  556. return nil
  557. }
  558. // Read methods
  559. func (c *Conn) advanceFrame() (int, error) {
  560. // 1. Skip remainder of previous frame.
  561. if c.readRemaining > 0 {
  562. if _, err := io.CopyN(ioutil.Discard, c.br, c.readRemaining); err != nil {
  563. return noFrame, err
  564. }
  565. }
  566. // 2. Read and parse first two bytes of frame header.
  567. p, err := c.read(2)
  568. if err != nil {
  569. return noFrame, err
  570. }
  571. final := p[0]&finalBit != 0
  572. frameType := int(p[0] & 0xf)
  573. reserved := int((p[0] >> 4) & 0x7)
  574. mask := p[1]&maskBit != 0
  575. c.readRemaining = int64(p[1] & 0x7f)
  576. if reserved != 0 {
  577. return noFrame, c.handleProtocolError("unexpected reserved bits " + strconv.Itoa(reserved))
  578. }
  579. switch frameType {
  580. case CloseMessage, PingMessage, PongMessage:
  581. if c.readRemaining > maxControlFramePayloadSize {
  582. return noFrame, c.handleProtocolError("control frame length > 125")
  583. }
  584. if !final {
  585. return noFrame, c.handleProtocolError("control frame not final")
  586. }
  587. case TextMessage, BinaryMessage:
  588. if !c.readFinal {
  589. return noFrame, c.handleProtocolError("message start before final message frame")
  590. }
  591. c.readFinal = final
  592. case continuationFrame:
  593. if c.readFinal {
  594. return noFrame, c.handleProtocolError("continuation after final message frame")
  595. }
  596. c.readFinal = final
  597. default:
  598. return noFrame, c.handleProtocolError("unknown opcode " + strconv.Itoa(frameType))
  599. }
  600. // 3. Read and parse frame length.
  601. switch c.readRemaining {
  602. case 126:
  603. p, err := c.read(2)
  604. if err != nil {
  605. return noFrame, err
  606. }
  607. c.readRemaining = int64(binary.BigEndian.Uint16(p))
  608. case 127:
  609. p, err := c.read(8)
  610. if err != nil {
  611. return noFrame, err
  612. }
  613. c.readRemaining = int64(binary.BigEndian.Uint64(p))
  614. }
  615. // 4. Handle frame masking.
  616. if mask != c.isServer {
  617. return noFrame, c.handleProtocolError("incorrect mask flag")
  618. }
  619. if mask {
  620. c.readMaskPos = 0
  621. p, err := c.read(len(c.readMaskKey))
  622. if err != nil {
  623. return noFrame, err
  624. }
  625. copy(c.readMaskKey[:], p)
  626. }
  627. // 5. For text and binary messages, enforce read limit and return.
  628. if frameType == continuationFrame || frameType == TextMessage || frameType == BinaryMessage {
  629. c.readLength += c.readRemaining
  630. if c.readLimit > 0 && c.readLength > c.readLimit {
  631. c.WriteControl(CloseMessage, FormatCloseMessage(CloseMessageTooBig, ""), time.Now().Add(writeWait))
  632. return noFrame, ErrReadLimit
  633. }
  634. return frameType, nil
  635. }
  636. // 6. Read control frame payload.
  637. var payload []byte
  638. if c.readRemaining > 0 {
  639. payload, err = c.read(int(c.readRemaining))
  640. c.readRemaining = 0
  641. if err != nil {
  642. return noFrame, err
  643. }
  644. if c.isServer {
  645. maskBytes(c.readMaskKey, 0, payload)
  646. }
  647. }
  648. // 7. Process control frame payload.
  649. switch frameType {
  650. case PongMessage:
  651. if err := c.handlePong(string(payload)); err != nil {
  652. return noFrame, err
  653. }
  654. case PingMessage:
  655. if err := c.handlePing(string(payload)); err != nil {
  656. return noFrame, err
  657. }
  658. case CloseMessage:
  659. echoMessage := []byte{}
  660. closeCode := CloseNoStatusReceived
  661. closeText := ""
  662. if len(payload) >= 2 {
  663. echoMessage = payload[:2]
  664. closeCode = int(binary.BigEndian.Uint16(payload))
  665. if !isValidReceivedCloseCode(closeCode) {
  666. return noFrame, c.handleProtocolError("invalid close code")
  667. }
  668. closeText = string(payload[2:])
  669. if !utf8.ValidString(closeText) {
  670. return noFrame, c.handleProtocolError("invalid utf8 payload in close frame")
  671. }
  672. }
  673. c.WriteControl(CloseMessage, echoMessage, time.Now().Add(writeWait))
  674. return noFrame, &CloseError{Code: closeCode, Text: closeText}
  675. }
  676. return frameType, nil
  677. }
  678. func (c *Conn) handleProtocolError(message string) error {
  679. c.WriteControl(CloseMessage, FormatCloseMessage(CloseProtocolError, message), time.Now().Add(writeWait))
  680. return errors.New("websocket: " + message)
  681. }
  682. // NextReader returns the next data message received from the peer. The
  683. // returned messageType is either TextMessage or BinaryMessage.
  684. //
  685. // There can be at most one open reader on a connection. NextReader discards
  686. // the previous message if the application has not already consumed it.
  687. //
  688. // Applications must break out of the application's read loop when this method
  689. // returns a non-nil error value. Errors returned from this method are
  690. // permanent. Once this method returns a non-nil error, all subsequent calls to
  691. // this method return the same error.
  692. func (c *Conn) NextReader() (messageType int, r io.Reader, err error) {
  693. c.messageReader = nil
  694. c.readLength = 0
  695. for c.readErr == nil {
  696. frameType, err := c.advanceFrame()
  697. if err != nil {
  698. c.readErr = hideTempErr(err)
  699. break
  700. }
  701. if frameType == TextMessage || frameType == BinaryMessage {
  702. r := &messageReader{c}
  703. c.messageReader = r
  704. return frameType, r, nil
  705. }
  706. }
  707. // Applications that do handle the error returned from this method spin in
  708. // tight loop on connection failure. To help application developers detect
  709. // this error, panic on repeated reads to the failed connection.
  710. c.readErrCount++
  711. if c.readErrCount >= 1000 {
  712. panic("repeated read on failed websocket connection")
  713. }
  714. return noFrame, nil, c.readErr
  715. }
  716. type messageReader struct{ c *Conn }
  717. func (r *messageReader) Read(b []byte) (int, error) {
  718. c := r.c
  719. if c.messageReader != r {
  720. return 0, io.EOF
  721. }
  722. for c.readErr == nil {
  723. if c.readRemaining > 0 {
  724. if int64(len(b)) > c.readRemaining {
  725. b = b[:c.readRemaining]
  726. }
  727. n, err := c.br.Read(b)
  728. c.readErr = hideTempErr(err)
  729. if c.isServer {
  730. c.readMaskPos = maskBytes(c.readMaskKey, c.readMaskPos, b[:n])
  731. }
  732. c.readRemaining -= int64(n)
  733. if c.readRemaining > 0 && c.readErr == io.EOF {
  734. c.readErr = errUnexpectedEOF
  735. }
  736. return n, c.readErr
  737. }
  738. if c.readFinal {
  739. c.messageReader = nil
  740. return 0, io.EOF
  741. }
  742. frameType, err := c.advanceFrame()
  743. switch {
  744. case err != nil:
  745. c.readErr = hideTempErr(err)
  746. case frameType == TextMessage || frameType == BinaryMessage:
  747. c.readErr = errors.New("websocket: internal error, unexpected text or binary in Reader")
  748. }
  749. }
  750. err := c.readErr
  751. if err == io.EOF && c.messageReader == r {
  752. err = errUnexpectedEOF
  753. }
  754. return 0, err
  755. }
  756. // ReadMessage is a helper method for getting a reader using NextReader and
  757. // reading from that reader to a buffer.
  758. func (c *Conn) ReadMessage() (messageType int, p []byte, err error) {
  759. var r io.Reader
  760. messageType, r, err = c.NextReader()
  761. if err != nil {
  762. return messageType, nil, err
  763. }
  764. p, err = ioutil.ReadAll(r)
  765. return messageType, p, err
  766. }
  767. // SetReadDeadline sets the read deadline on the underlying network connection.
  768. // After a read has timed out, the websocket connection state is corrupt and
  769. // all future reads will return an error. A zero value for t means reads will
  770. // not time out.
  771. func (c *Conn) SetReadDeadline(t time.Time) error {
  772. return c.conn.SetReadDeadline(t)
  773. }
  774. // SetReadLimit sets the maximum size for a message read from the peer. If a
  775. // message exceeds the limit, the connection sends a close frame to the peer
  776. // and returns ErrReadLimit to the application.
  777. func (c *Conn) SetReadLimit(limit int64) {
  778. c.readLimit = limit
  779. }
  780. // PingHandler returns the current ping handler
  781. func (c *Conn) PingHandler() func(appData string) error {
  782. return c.handlePing
  783. }
  784. // SetPingHandler sets the handler for ping messages received from the peer.
  785. // The appData argument to h is the PING frame application data. The default
  786. // ping handler sends a pong to the peer.
  787. func (c *Conn) SetPingHandler(h func(appData string) error) {
  788. if h == nil {
  789. h = func(message string) error {
  790. err := c.WriteControl(PongMessage, []byte(message), time.Now().Add(writeWait))
  791. if err == ErrCloseSent {
  792. return nil
  793. } else if e, ok := err.(net.Error); ok && e.Temporary() {
  794. return nil
  795. }
  796. return err
  797. }
  798. }
  799. c.handlePing = h
  800. }
  801. // PongHandler returns the current pong handler
  802. func (c *Conn) PongHandler() func(appData string) error {
  803. return c.handlePong
  804. }
  805. // SetPongHandler sets the handler for pong messages received from the peer.
  806. // The appData argument to h is the PONG frame application data. The default
  807. // pong handler does nothing.
  808. func (c *Conn) SetPongHandler(h func(appData string) error) {
  809. if h == nil {
  810. h = func(string) error { return nil }
  811. }
  812. c.handlePong = h
  813. }
  814. // UnderlyingConn returns the internal net.Conn. This can be used to further
  815. // modifications to connection specific flags.
  816. func (c *Conn) UnderlyingConn() net.Conn {
  817. return c.conn
  818. }
  819. // FormatCloseMessage formats closeCode and text as a WebSocket close message.
  820. func FormatCloseMessage(closeCode int, text string) []byte {
  821. buf := make([]byte, 2+len(text))
  822. binary.BigEndian.PutUint16(buf, uint16(closeCode))
  823. copy(buf[2:], text)
  824. return buf
  825. }