conn.go 23 KB

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