conn.go 24 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950
  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. writeSeq int // incremented to invalidate message writers.
  209. writeDeadline time.Time
  210. isWriting bool // for best-effort concurrent write detection
  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. readSeq int // incremented to invalidate message readers.
  217. readLength int64 // Message size.
  218. readLimit int64 // Maximum message size.
  219. readMaskPos int
  220. readMaskKey [4]byte
  221. handlePong func(string) error
  222. handlePing func(string) error
  223. readErrCount int
  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 writeBufferSize == 0 {
  232. writeBufferSize = defaultWriteBufferSize
  233. }
  234. c := &Conn{
  235. isServer: isServer,
  236. br: bufio.NewReaderSize(conn, readBufferSize),
  237. conn: conn,
  238. mu: mu,
  239. readFinal: true,
  240. writeBuf: make([]byte, writeBufferSize+maxFrameHeaderSize),
  241. writeFrameType: noFrame,
  242. writePos: maxFrameHeaderSize,
  243. }
  244. c.SetPingHandler(nil)
  245. c.SetPongHandler(nil)
  246. return c
  247. }
  248. // Subprotocol returns the negotiated protocol for the connection.
  249. func (c *Conn) Subprotocol() string {
  250. return c.subprotocol
  251. }
  252. // Close closes the underlying network connection without sending or waiting for a close frame.
  253. func (c *Conn) Close() error {
  254. return c.conn.Close()
  255. }
  256. // LocalAddr returns the local network address.
  257. func (c *Conn) LocalAddr() net.Addr {
  258. return c.conn.LocalAddr()
  259. }
  260. // RemoteAddr returns the remote network address.
  261. func (c *Conn) RemoteAddr() net.Addr {
  262. return c.conn.RemoteAddr()
  263. }
  264. // Write methods
  265. func (c *Conn) write(frameType int, deadline time.Time, bufs ...[]byte) error {
  266. <-c.mu
  267. defer func() { c.mu <- true }()
  268. if c.closeSent {
  269. return ErrCloseSent
  270. } else if frameType == CloseMessage {
  271. c.closeSent = true
  272. }
  273. c.conn.SetWriteDeadline(deadline)
  274. for _, buf := range bufs {
  275. if len(buf) > 0 {
  276. n, err := c.conn.Write(buf)
  277. if n != len(buf) {
  278. // Close on partial write.
  279. c.conn.Close()
  280. }
  281. if err != nil {
  282. return err
  283. }
  284. }
  285. }
  286. return nil
  287. }
  288. // WriteControl writes a control message with the given deadline. The allowed
  289. // message types are CloseMessage, PingMessage and PongMessage.
  290. func (c *Conn) WriteControl(messageType int, data []byte, deadline time.Time) error {
  291. if !isControl(messageType) {
  292. return errBadWriteOpCode
  293. }
  294. if len(data) > maxControlFramePayloadSize {
  295. return errInvalidControlFrame
  296. }
  297. b0 := byte(messageType) | finalBit
  298. b1 := byte(len(data))
  299. if !c.isServer {
  300. b1 |= maskBit
  301. }
  302. buf := make([]byte, 0, maxFrameHeaderSize+maxControlFramePayloadSize)
  303. buf = append(buf, b0, b1)
  304. if c.isServer {
  305. buf = append(buf, data...)
  306. } else {
  307. key := newMaskKey()
  308. buf = append(buf, key[:]...)
  309. buf = append(buf, data...)
  310. maskBytes(key, 0, buf[6:])
  311. }
  312. d := time.Hour * 1000
  313. if !deadline.IsZero() {
  314. d = deadline.Sub(time.Now())
  315. if d < 0 {
  316. return errWriteTimeout
  317. }
  318. }
  319. timer := time.NewTimer(d)
  320. select {
  321. case <-c.mu:
  322. timer.Stop()
  323. case <-timer.C:
  324. return errWriteTimeout
  325. }
  326. defer func() { c.mu <- true }()
  327. if c.closeSent {
  328. return ErrCloseSent
  329. } else if messageType == CloseMessage {
  330. c.closeSent = true
  331. }
  332. c.conn.SetWriteDeadline(deadline)
  333. n, err := c.conn.Write(buf)
  334. if n != 0 && n != len(buf) {
  335. c.conn.Close()
  336. }
  337. return hideTempErr(err)
  338. }
  339. // NextWriter returns a writer for the next message to send. The writer's
  340. // Close method flushes the complete message to the network.
  341. //
  342. // There can be at most one open writer on a connection. NextWriter closes the
  343. // previous writer if the application has not already done so.
  344. func (c *Conn) NextWriter(messageType int) (io.WriteCloser, error) {
  345. if c.writeErr != nil {
  346. return nil, c.writeErr
  347. }
  348. if c.writeFrameType != noFrame {
  349. if err := c.flushFrame(true, nil); err != nil {
  350. return nil, err
  351. }
  352. }
  353. if !isControl(messageType) && !isData(messageType) {
  354. return nil, errBadWriteOpCode
  355. }
  356. c.writeFrameType = messageType
  357. return messageWriter{c, c.writeSeq}, nil
  358. }
  359. func (c *Conn) flushFrame(final bool, extra []byte) error {
  360. length := c.writePos - maxFrameHeaderSize + len(extra)
  361. // Check for invalid control frames.
  362. if isControl(c.writeFrameType) &&
  363. (!final || length > maxControlFramePayloadSize) {
  364. c.writeSeq++
  365. c.writeFrameType = noFrame
  366. c.writePos = maxFrameHeaderSize
  367. return errInvalidControlFrame
  368. }
  369. b0 := byte(c.writeFrameType)
  370. if final {
  371. b0 |= finalBit
  372. }
  373. b1 := byte(0)
  374. if !c.isServer {
  375. b1 |= maskBit
  376. }
  377. // Assume that the frame starts at beginning of c.writeBuf.
  378. framePos := 0
  379. if c.isServer {
  380. // Adjust up if mask not included in the header.
  381. framePos = 4
  382. }
  383. switch {
  384. case length >= 65536:
  385. c.writeBuf[framePos] = b0
  386. c.writeBuf[framePos+1] = b1 | 127
  387. binary.BigEndian.PutUint64(c.writeBuf[framePos+2:], uint64(length))
  388. case length > 125:
  389. framePos += 6
  390. c.writeBuf[framePos] = b0
  391. c.writeBuf[framePos+1] = b1 | 126
  392. binary.BigEndian.PutUint16(c.writeBuf[framePos+2:], uint16(length))
  393. default:
  394. framePos += 8
  395. c.writeBuf[framePos] = b0
  396. c.writeBuf[framePos+1] = b1 | byte(length)
  397. }
  398. if !c.isServer {
  399. key := newMaskKey()
  400. copy(c.writeBuf[maxFrameHeaderSize-4:], key[:])
  401. maskBytes(key, 0, c.writeBuf[maxFrameHeaderSize:c.writePos])
  402. if len(extra) > 0 {
  403. c.writeErr = errors.New("websocket: internal error, extra used in client mode")
  404. return c.writeErr
  405. }
  406. }
  407. // Write the buffers to the connection with best-effort detection of
  408. // concurrent writes. See the concurrency section in the package
  409. // documentation for more info.
  410. if c.isWriting {
  411. panic("concurrent write to websocket connection")
  412. }
  413. c.isWriting = true
  414. c.writeErr = c.write(c.writeFrameType, c.writeDeadline, c.writeBuf[framePos:c.writePos], extra)
  415. if !c.isWriting {
  416. panic("concurrent write to websocket connection")
  417. }
  418. c.isWriting = false
  419. // Setup for next frame.
  420. c.writePos = maxFrameHeaderSize
  421. c.writeFrameType = continuationFrame
  422. if final {
  423. c.writeSeq++
  424. c.writeFrameType = noFrame
  425. }
  426. return c.writeErr
  427. }
  428. type messageWriter struct {
  429. c *Conn
  430. seq int
  431. }
  432. func (w messageWriter) err() error {
  433. c := w.c
  434. if c.writeSeq != w.seq {
  435. return errWriteClosed
  436. }
  437. if c.writeErr != nil {
  438. return c.writeErr
  439. }
  440. return nil
  441. }
  442. func (w messageWriter) ncopy(max int) (int, error) {
  443. n := len(w.c.writeBuf) - w.c.writePos
  444. if n <= 0 {
  445. if err := w.c.flushFrame(false, nil); err != nil {
  446. return 0, err
  447. }
  448. n = len(w.c.writeBuf) - w.c.writePos
  449. }
  450. if n > max {
  451. n = max
  452. }
  453. return n, nil
  454. }
  455. func (w messageWriter) write(final bool, p []byte) (int, error) {
  456. if err := w.err(); err != nil {
  457. return 0, err
  458. }
  459. if len(p) > 2*len(w.c.writeBuf) && w.c.isServer {
  460. // Don't buffer large messages.
  461. err := w.c.flushFrame(final, p)
  462. if err != nil {
  463. return 0, err
  464. }
  465. return len(p), nil
  466. }
  467. nn := len(p)
  468. for len(p) > 0 {
  469. n, err := w.ncopy(len(p))
  470. if err != nil {
  471. return 0, err
  472. }
  473. copy(w.c.writeBuf[w.c.writePos:], p[:n])
  474. w.c.writePos += n
  475. p = p[n:]
  476. }
  477. return nn, nil
  478. }
  479. func (w messageWriter) Write(p []byte) (int, error) {
  480. return w.write(false, p)
  481. }
  482. func (w messageWriter) WriteString(p string) (int, error) {
  483. if err := w.err(); err != nil {
  484. return 0, err
  485. }
  486. nn := len(p)
  487. for len(p) > 0 {
  488. n, err := w.ncopy(len(p))
  489. if err != nil {
  490. return 0, err
  491. }
  492. copy(w.c.writeBuf[w.c.writePos:], p[:n])
  493. w.c.writePos += n
  494. p = p[n:]
  495. }
  496. return nn, nil
  497. }
  498. func (w messageWriter) ReadFrom(r io.Reader) (nn int64, err error) {
  499. if err := w.err(); err != nil {
  500. return 0, err
  501. }
  502. for {
  503. if w.c.writePos == len(w.c.writeBuf) {
  504. err = w.c.flushFrame(false, nil)
  505. if err != nil {
  506. break
  507. }
  508. }
  509. var n int
  510. n, err = r.Read(w.c.writeBuf[w.c.writePos:])
  511. w.c.writePos += n
  512. nn += int64(n)
  513. if err != nil {
  514. if err == io.EOF {
  515. err = nil
  516. }
  517. break
  518. }
  519. }
  520. return nn, err
  521. }
  522. func (w messageWriter) Close() error {
  523. if err := w.err(); err != nil {
  524. return err
  525. }
  526. return w.c.flushFrame(true, nil)
  527. }
  528. // WriteMessage is a helper method for getting a writer using NextWriter,
  529. // writing the message and closing the writer.
  530. func (c *Conn) WriteMessage(messageType int, data []byte) error {
  531. wr, err := c.NextWriter(messageType)
  532. if err != nil {
  533. return err
  534. }
  535. w := wr.(messageWriter)
  536. if _, err := w.write(true, data); err != nil {
  537. return err
  538. }
  539. if c.writeSeq == w.seq {
  540. if err := c.flushFrame(true, nil); err != nil {
  541. return err
  542. }
  543. }
  544. return nil
  545. }
  546. // SetWriteDeadline sets the write deadline on the underlying network
  547. // connection. After a write has timed out, the websocket state is corrupt and
  548. // all future writes will return an error. A zero value for t means writes will
  549. // not time out.
  550. func (c *Conn) SetWriteDeadline(t time.Time) error {
  551. c.writeDeadline = t
  552. return nil
  553. }
  554. // Read methods
  555. // readFull is like io.ReadFull except that io.EOF is never returned.
  556. func (c *Conn) readFull(p []byte) (err error) {
  557. var n int
  558. for n < len(p) && err == nil {
  559. var nn int
  560. nn, err = c.br.Read(p[n:])
  561. n += nn
  562. }
  563. if n == len(p) {
  564. err = nil
  565. } else if err == io.EOF {
  566. err = errUnexpectedEOF
  567. }
  568. return
  569. }
  570. func (c *Conn) advanceFrame() (int, error) {
  571. // 1. Skip remainder of previous frame.
  572. if c.readRemaining > 0 {
  573. if _, err := io.CopyN(ioutil.Discard, c.br, c.readRemaining); err != nil {
  574. return noFrame, err
  575. }
  576. }
  577. // 2. Read and parse first two bytes of frame header.
  578. var b [8]byte
  579. if err := c.readFull(b[:2]); err != nil {
  580. return noFrame, err
  581. }
  582. final := b[0]&finalBit != 0
  583. frameType := int(b[0] & 0xf)
  584. reserved := int((b[0] >> 4) & 0x7)
  585. mask := b[1]&maskBit != 0
  586. c.readRemaining = int64(b[1] & 0x7f)
  587. if reserved != 0 {
  588. return noFrame, c.handleProtocolError("unexpected reserved bits " + strconv.Itoa(reserved))
  589. }
  590. switch frameType {
  591. case CloseMessage, PingMessage, PongMessage:
  592. if c.readRemaining > maxControlFramePayloadSize {
  593. return noFrame, c.handleProtocolError("control frame length > 125")
  594. }
  595. if !final {
  596. return noFrame, c.handleProtocolError("control frame not final")
  597. }
  598. case TextMessage, BinaryMessage:
  599. if !c.readFinal {
  600. return noFrame, c.handleProtocolError("message start before final message frame")
  601. }
  602. c.readFinal = final
  603. case continuationFrame:
  604. if c.readFinal {
  605. return noFrame, c.handleProtocolError("continuation after final message frame")
  606. }
  607. c.readFinal = final
  608. default:
  609. return noFrame, c.handleProtocolError("unknown opcode " + strconv.Itoa(frameType))
  610. }
  611. // 3. Read and parse frame length.
  612. switch c.readRemaining {
  613. case 126:
  614. if err := c.readFull(b[:2]); err != nil {
  615. return noFrame, err
  616. }
  617. c.readRemaining = int64(binary.BigEndian.Uint16(b[:2]))
  618. case 127:
  619. if err := c.readFull(b[:8]); err != nil {
  620. return noFrame, err
  621. }
  622. c.readRemaining = int64(binary.BigEndian.Uint64(b[:8]))
  623. }
  624. // 4. Handle frame masking.
  625. if mask != c.isServer {
  626. return noFrame, c.handleProtocolError("incorrect mask flag")
  627. }
  628. if mask {
  629. c.readMaskPos = 0
  630. if err := c.readFull(c.readMaskKey[:]); err != nil {
  631. return noFrame, err
  632. }
  633. }
  634. // 5. For text and binary messages, enforce read limit and return.
  635. if frameType == continuationFrame || frameType == TextMessage || frameType == BinaryMessage {
  636. c.readLength += c.readRemaining
  637. if c.readLimit > 0 && c.readLength > c.readLimit {
  638. c.WriteControl(CloseMessage, FormatCloseMessage(CloseMessageTooBig, ""), time.Now().Add(writeWait))
  639. return noFrame, ErrReadLimit
  640. }
  641. return frameType, nil
  642. }
  643. // 6. Read control frame payload.
  644. var payload []byte
  645. if c.readRemaining > 0 {
  646. payload = make([]byte, c.readRemaining)
  647. c.readRemaining = 0
  648. if err := c.readFull(payload); err != nil {
  649. return noFrame, err
  650. }
  651. if c.isServer {
  652. maskBytes(c.readMaskKey, 0, payload)
  653. }
  654. }
  655. // 7. Process control frame payload.
  656. switch frameType {
  657. case PongMessage:
  658. if err := c.handlePong(string(payload)); err != nil {
  659. return noFrame, err
  660. }
  661. case PingMessage:
  662. if err := c.handlePing(string(payload)); err != nil {
  663. return noFrame, err
  664. }
  665. case CloseMessage:
  666. echoMessage := []byte{}
  667. closeCode := CloseNoStatusReceived
  668. closeText := ""
  669. if len(payload) >= 2 {
  670. echoMessage = payload[:2]
  671. closeCode = int(binary.BigEndian.Uint16(payload))
  672. if !isValidReceivedCloseCode(closeCode) {
  673. return noFrame, c.handleProtocolError("invalid close code")
  674. }
  675. closeText = string(payload[2:])
  676. if !utf8.ValidString(closeText) {
  677. return noFrame, c.handleProtocolError("invalid utf8 payload in close frame")
  678. }
  679. }
  680. c.WriteControl(CloseMessage, echoMessage, time.Now().Add(writeWait))
  681. return noFrame, &CloseError{Code: closeCode, Text: closeText}
  682. }
  683. return frameType, nil
  684. }
  685. func (c *Conn) handleProtocolError(message string) error {
  686. c.WriteControl(CloseMessage, FormatCloseMessage(CloseProtocolError, message), time.Now().Add(writeWait))
  687. return errors.New("websocket: " + message)
  688. }
  689. // NextReader returns the next data message received from the peer. The
  690. // returned messageType is either TextMessage or BinaryMessage.
  691. //
  692. // There can be at most one open reader on a connection. NextReader discards
  693. // the previous message if the application has not already consumed it.
  694. //
  695. // Applications must break out of the application's read loop when this method
  696. // returns a non-nil error value. Errors returned from this method are
  697. // permanent. Once this method returns a non-nil error, all subsequent calls to
  698. // this method return the same error.
  699. func (c *Conn) NextReader() (messageType int, r io.Reader, err error) {
  700. c.readSeq++
  701. c.readLength = 0
  702. for c.readErr == nil {
  703. frameType, err := c.advanceFrame()
  704. if err != nil {
  705. c.readErr = hideTempErr(err)
  706. break
  707. }
  708. if frameType == TextMessage || frameType == BinaryMessage {
  709. return frameType, messageReader{c, c.readSeq}, nil
  710. }
  711. }
  712. // Applications that do handle the error returned from this method spin in
  713. // tight loop on connection failure. To help application developers detect
  714. // this error, panic on repeated reads to the failed connection.
  715. c.readErrCount++
  716. if c.readErrCount >= 1000 {
  717. panic("repeated read on failed websocket connection")
  718. }
  719. return noFrame, nil, c.readErr
  720. }
  721. type messageReader struct {
  722. c *Conn
  723. seq int
  724. }
  725. func (r messageReader) Read(b []byte) (int, error) {
  726. if r.seq != r.c.readSeq {
  727. return 0, io.EOF
  728. }
  729. for r.c.readErr == nil {
  730. if r.c.readRemaining > 0 {
  731. if int64(len(b)) > r.c.readRemaining {
  732. b = b[:r.c.readRemaining]
  733. }
  734. n, err := r.c.br.Read(b)
  735. r.c.readErr = hideTempErr(err)
  736. if r.c.isServer {
  737. r.c.readMaskPos = maskBytes(r.c.readMaskKey, r.c.readMaskPos, b[:n])
  738. }
  739. r.c.readRemaining -= int64(n)
  740. if r.c.readRemaining > 0 && r.c.readErr == io.EOF {
  741. r.c.readErr = errUnexpectedEOF
  742. }
  743. return n, r.c.readErr
  744. }
  745. if r.c.readFinal {
  746. r.c.readSeq++
  747. return 0, io.EOF
  748. }
  749. frameType, err := r.c.advanceFrame()
  750. switch {
  751. case err != nil:
  752. r.c.readErr = hideTempErr(err)
  753. case frameType == TextMessage || frameType == BinaryMessage:
  754. r.c.readErr = errors.New("websocket: internal error, unexpected text or binary in Reader")
  755. }
  756. }
  757. err := r.c.readErr
  758. if err == io.EOF && r.seq == r.c.readSeq {
  759. err = errUnexpectedEOF
  760. }
  761. return 0, err
  762. }
  763. // ReadMessage is a helper method for getting a reader using NextReader and
  764. // reading from that reader to a buffer.
  765. func (c *Conn) ReadMessage() (messageType int, p []byte, err error) {
  766. var r io.Reader
  767. messageType, r, err = c.NextReader()
  768. if err != nil {
  769. return messageType, nil, err
  770. }
  771. p, err = ioutil.ReadAll(r)
  772. return messageType, p, err
  773. }
  774. // SetReadDeadline sets the read deadline on the underlying network connection.
  775. // After a read has timed out, the websocket connection state is corrupt and
  776. // all future reads will return an error. A zero value for t means reads will
  777. // not time out.
  778. func (c *Conn) SetReadDeadline(t time.Time) error {
  779. return c.conn.SetReadDeadline(t)
  780. }
  781. // SetReadLimit sets the maximum size for a message read from the peer. If a
  782. // message exceeds the limit, the connection sends a close frame to the peer
  783. // and returns ErrReadLimit to the application.
  784. func (c *Conn) SetReadLimit(limit int64) {
  785. c.readLimit = limit
  786. }
  787. // SetPingHandler sets the handler for ping messages received from the peer.
  788. // The appData argument to h is the PING frame application data. The default
  789. // ping handler sends a pong to the peer.
  790. func (c *Conn) SetPingHandler(h func(appData string) error) {
  791. if h == nil {
  792. h = func(message string) error {
  793. err := c.WriteControl(PongMessage, []byte(message), time.Now().Add(writeWait))
  794. if err == ErrCloseSent {
  795. return nil
  796. } else if e, ok := err.(net.Error); ok && e.Temporary() {
  797. return nil
  798. }
  799. return err
  800. }
  801. }
  802. c.handlePing = h
  803. }
  804. // SetPongHandler sets the handler for pong messages received from the peer.
  805. // The appData argument to h is the PONG frame application data. The default
  806. // pong handler does nothing.
  807. func (c *Conn) SetPongHandler(h func(appData string) error) {
  808. if h == nil {
  809. h = func(string) error { return nil }
  810. }
  811. c.handlePong = h
  812. }
  813. // UnderlyingConn returns the internal net.Conn. This can be used to further
  814. // modifications to connection specific flags.
  815. func (c *Conn) UnderlyingConn() net.Conn {
  816. return c.conn
  817. }
  818. // FormatCloseMessage formats closeCode and text as a WebSocket close message.
  819. func FormatCloseMessage(closeCode int, text string) []byte {
  820. buf := make([]byte, 2+len(text))
  821. binary.BigEndian.PutUint16(buf, uint16(closeCode))
  822. copy(buf[2:], text)
  823. return buf
  824. }