(inChannels base.Set, options db.ChangesOptions)
| 512 | } |
| 513 | |
| 514 | func (h *handler) sendContinuousChangesByWebSocket(inChannels base.Set, options db.ChangesOptions) (error, bool) { |
| 515 | |
| 516 | forceClose := false |
| 517 | handler := func(conn *websocket.Conn) { |
| 518 | h.logStatus(101, "Upgraded to WebSocket protocol") |
| 519 | defer func() { |
| 520 | if err := conn.Close(); err != nil { |
| 521 | base.WarnfCtx(h.ctx(), "WebSocket connection (%s) closed with error %v", h.formatSerialNumber(), err) |
| 522 | } |
| 523 | base.InfofCtx(h.ctx(), base.KeyHTTP, "%s: --> WebSocket closed", h.formatSerialNumber()) |
| 524 | }() |
| 525 | |
| 526 | // Read changes-feed options from an initial incoming WebSocket message in JSON format: |
| 527 | var wsoptions db.ChangesOptions |
| 528 | var compress bool |
| 529 | if msg, err := readWebSocketMessage(h.ctx(), conn); err != nil { |
| 530 | return |
| 531 | } else { |
| 532 | var channelNames []string |
| 533 | var err error |
| 534 | if _, wsoptions, _, channelNames, _, compress, err = h.readChangesOptionsFromJSON(msg); err != nil { |
| 535 | return |
| 536 | } |
| 537 | if channelNames != nil { |
| 538 | inChannels, _ = ch.SetFromArray(channelNames, ch.ExpandStar) |
| 539 | } |
| 540 | } |
| 541 | |
| 542 | // Copy options.ChangesCtx to new WebSocket options |
| 543 | // options.ChangesCtx will be cancelled automatically when |
| 544 | // changes feed completes |
| 545 | wsoptions.ChangesCtx = options.ChangesCtx |
| 546 | |
| 547 | // Set up GZip compression |
| 548 | var writer *bytes.Buffer |
| 549 | var zipWriter *gzip.Writer |
| 550 | if compress { |
| 551 | writer = bytes.NewBuffer(nil) |
| 552 | zipWriter = GetGZipWriter(writer) |
| 553 | } |
| 554 | |
| 555 | caughtUp := false |
| 556 | _, forceClose = h.generateContinuousChanges(inChannels, wsoptions, func(changes []*db.ChangeEntry) error { |
| 557 | var data []byte |
| 558 | if changes != nil { |
| 559 | data, _ = base.JSONMarshal(changes) |
| 560 | } else if !caughtUp { |
| 561 | caughtUp = true |
| 562 | data, _ = base.JSONMarshal([]*db.ChangeEntry{}) |
| 563 | } else { |
| 564 | data = []byte{} |
| 565 | } |
| 566 | if compress && len(data) > 8 { |
| 567 | // Compress JSON, using same GZip context, and send as binary msg: |
| 568 | _, _ = zipWriter.Write(data) |
| 569 | _ = zipWriter.Flush() |
| 570 | data = writer.Bytes() |
| 571 | writer.Reset() |
no test coverage detected