mirror of
https://github.com/gravitational/teleport.git
synced 2026-09-21 05:55:42 +08:00
Update the web UI to stream session playback (#36168)
Prior to this, the web UI would download the entire session recording and store it in JavaScript memory before starting playback. This caused the browser tab to crash when attempting to play back sessions larger than ~5MB. For playback, we use a custom binary protocol rather than the protobuf envelopes that we use for live sessions. The protobuf envelopes only send raw PTY data, there is no place to put the timing data. Adding fields to the envelope would be a disruptive change because our JS codec is hand-rolled and we'd have to make the parsing updates manually. Updates gravitational/teleport-private#1024 Closes gravitational/teleport-private#665 Closes #10578
This commit is contained in:
Zac Bergquist
@@ -144,8 +144,8 @@ func NewAPIServer(config *APIConfig) (http.Handler, error) {
|
||||
srv.POST("/:version/tokens/register", srv.WithAuth(srv.registerUsingToken))
|
||||
|
||||
// Active sessions
|
||||
srv.GET("/:version/namespaces/:namespace/sessions/:id/stream", srv.WithAuth(srv.getSessionChunk))
|
||||
srv.GET("/:version/namespaces/:namespace/sessions/:id/events", srv.WithAuth(srv.getSessionEvents))
|
||||
srv.GET("/:version/namespaces/:namespace/sessions/:id/stream", srv.WithAuth(srv.getSessionChunk)) // DELETE IN 16(zmb3)
|
||||
srv.GET("/:version/namespaces/:namespace/sessions/:id/events", srv.WithAuth(srv.getSessionEvents)) // DELETE IN 16(zmb3)
|
||||
|
||||
// Namespaces
|
||||
srv.POST("/:version/namespaces", srv.WithAuth(srv.upsertNamespace))
|
||||
|
||||
@@ -116,7 +116,7 @@ func New(cfg *Config) (*Player, error) {
|
||||
log: log,
|
||||
sessionID: cfg.SessionID,
|
||||
streamer: cfg.Streamer,
|
||||
emit: make(chan events.AuditEvent, 64),
|
||||
emit: make(chan events.AuditEvent, 1024),
|
||||
playPause: make(chan chan struct{}, 1),
|
||||
done: make(chan struct{}),
|
||||
}
|
||||
@@ -185,7 +185,7 @@ func (p *Player) stream() {
|
||||
}
|
||||
|
||||
currentDelay := getDelay(evt)
|
||||
if currentDelay > 0 && currentDelay > lastDelay {
|
||||
if currentDelay > 0 && currentDelay >= lastDelay {
|
||||
switch adv := p.advanceTo.Load(); {
|
||||
case adv >= currentDelay:
|
||||
// no timing delay necessary, we are fast forwarding
|
||||
@@ -215,12 +215,13 @@ func (p *Player) stream() {
|
||||
lastDelay = currentDelay
|
||||
}
|
||||
|
||||
select {
|
||||
case p.emit <- evt:
|
||||
p.lastPlayed.Store(currentDelay)
|
||||
default:
|
||||
p.log.Warnf("dropped event %v, reader too slow", evt.GetID())
|
||||
}
|
||||
// if the receiver can't keep up, let the channel throttle us
|
||||
// (it's better for playback to be a little slower than realtime
|
||||
// than to drop events)
|
||||
//
|
||||
// TODO: consider a select with a timeout to detect blocked readers?
|
||||
p.emit <- evt
|
||||
p.lastPlayed.Store(currentDelay)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -169,6 +169,7 @@ func TestClose(t *testing.T) {
|
||||
_, ok := <-p.C()
|
||||
require.False(t, ok, "player channel should have been closed")
|
||||
require.NoError(t, p.Err())
|
||||
require.Equal(t, int64(1000), p.LastPlayed())
|
||||
}
|
||||
|
||||
func TestSeekForward(t *testing.T) {
|
||||
|
||||
@@ -717,8 +717,11 @@ func (h *Handler) bindDefaultEndpoints() {
|
||||
// Audit events handlers.
|
||||
h.GET("/webapi/sites/:site/events/search", h.WithClusterAuth(h.clusterSearchEvents)) // search site events
|
||||
h.GET("/webapi/sites/:site/events/search/sessions", h.WithClusterAuth(h.clusterSearchSessionEvents)) // search site session events
|
||||
h.GET("/webapi/sites/:site/sessions/:sid/events", h.WithClusterAuth(h.siteSessionEventsGet)) // get recorded session's timing information (from events)
|
||||
h.GET("/webapi/sites/:site/sessions/:sid/stream", h.siteSessionStreamGet) // get recorded session's bytes (from events)
|
||||
h.GET("/webapi/sites/:site/ttyplayback/:sid", h.WithClusterAuth(h.ttyPlaybackHandle))
|
||||
|
||||
// DELETE in 16(zmb3): v15+ web UIs use new streaming 'ttyplayback' endpoint
|
||||
h.GET("/webapi/sites/:site/sessions/:sid/events", h.WithClusterAuth(h.siteSessionEventsGet)) // get recorded session's timing information (from events)
|
||||
h.GET("/webapi/sites/:site/sessions/:sid/stream", h.siteSessionStreamGet) // get recorded session's bytes (from events)
|
||||
|
||||
// scp file transfer
|
||||
h.GET("/webapi/sites/:site/nodes/:server/:login/scp", h.WithClusterAuth(h.transferFile))
|
||||
@@ -3417,6 +3420,8 @@ func queryOrder(query url.Values, name string, def types.EventOrder) (types.Even
|
||||
// It returns the binary stream unencoded, directly in the respose body,
|
||||
// with Content-Type of application/octet-stream, gzipped with up to 95%
|
||||
// compression ratio.
|
||||
//
|
||||
// DELETE IN 16(zmb3)
|
||||
func (h *Handler) siteSessionStreamGet(w http.ResponseWriter, r *http.Request, p httprouter.Params) {
|
||||
httplib.SetNoCacheHeaders(w.Header())
|
||||
|
||||
|
||||
@@ -0,0 +1,358 @@
|
||||
/*
|
||||
* Teleport
|
||||
* Copyright (C) 2023 Gravitational, Inc.
|
||||
*
|
||||
* This program is free software: you can redistribute it and/or modify
|
||||
* it under the terms of the GNU Affero General Public License as published by
|
||||
* the Free Software Foundation, either version 3 of the License, or
|
||||
* (at your option) any later version.
|
||||
*
|
||||
* This program is distributed in the hope that it will be useful,
|
||||
* but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
* GNU Affero General Public License for more details.
|
||||
*
|
||||
* You should have received a copy of the GNU Affero General Public License
|
||||
* along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
*/
|
||||
|
||||
package web
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/binary"
|
||||
"net/http"
|
||||
"time"
|
||||
|
||||
"github.com/gorilla/websocket"
|
||||
"github.com/gravitational/trace"
|
||||
"github.com/julienschmidt/httprouter"
|
||||
|
||||
"github.com/gravitational/teleport/api/types/events"
|
||||
"github.com/gravitational/teleport/lib/player"
|
||||
"github.com/gravitational/teleport/lib/reversetunnelclient"
|
||||
"github.com/gravitational/teleport/lib/session"
|
||||
"github.com/gravitational/teleport/lib/utils"
|
||||
)
|
||||
|
||||
const (
|
||||
messageTypePTY = byte(1)
|
||||
messageTypeError = byte(2)
|
||||
messageTypePlayPause = byte(3)
|
||||
messageTypeSeek = byte(4)
|
||||
messageTypeResize = byte(5)
|
||||
)
|
||||
|
||||
const (
|
||||
severityError = byte(1)
|
||||
)
|
||||
|
||||
const (
|
||||
actionPlay = byte(0)
|
||||
actionPause = byte(1)
|
||||
)
|
||||
|
||||
func (h *Handler) ttyPlaybackHandle(
|
||||
w http.ResponseWriter,
|
||||
r *http.Request,
|
||||
p httprouter.Params,
|
||||
sctx *SessionContext,
|
||||
site reversetunnelclient.RemoteSite,
|
||||
) (interface{}, error) {
|
||||
sID := p.ByName("sid")
|
||||
if sID == "" {
|
||||
return nil, trace.BadParameter("missing session ID in request URL")
|
||||
}
|
||||
|
||||
clt, err := sctx.GetUserClient(r.Context(), site)
|
||||
if err != nil {
|
||||
return nil, trace.Wrap(err)
|
||||
}
|
||||
|
||||
h.log.Debug("upgrading to websocket")
|
||||
upgrader := websocket.Upgrader{
|
||||
ReadBufferSize: 1024,
|
||||
WriteBufferSize: 1024,
|
||||
}
|
||||
ws, err := upgrader.Upgrade(w, r, nil)
|
||||
if err != nil {
|
||||
h.log.Warn("failed upgrade", err)
|
||||
// if Upgrade fails, it automatically replies with an HTTP error
|
||||
// (this means we don't need to return an error here)
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
player, err := player.New(&player.Config{
|
||||
Clock: h.clock,
|
||||
Log: h.log,
|
||||
SessionID: session.ID(sID),
|
||||
Streamer: clt,
|
||||
})
|
||||
if err != nil {
|
||||
h.log.Warn("player error", err)
|
||||
writeError(ws, err)
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithCancel(r.Context())
|
||||
defer cancel()
|
||||
|
||||
go func() {
|
||||
defer cancel()
|
||||
for {
|
||||
typ, b, err := ws.ReadMessage()
|
||||
if err != nil {
|
||||
if !utils.IsOKNetworkError(err) {
|
||||
log.Warnf("websocket read error: %v", err)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
if typ != websocket.BinaryMessage {
|
||||
log.Debugf("skipping unknown websocket message type %v", typ)
|
||||
continue
|
||||
}
|
||||
|
||||
if err := handlePlaybackAction(b, player); err != nil {
|
||||
log.Warnf("skipping bad action: %v", err)
|
||||
continue
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
go func() {
|
||||
defer cancel()
|
||||
defer func() {
|
||||
h.log.Debug("closing websocket")
|
||||
if err := ws.WriteMessage(websocket.CloseMessage, nil); err != nil {
|
||||
h.log.Debugf("error sending close message: %v", err)
|
||||
}
|
||||
if err := ws.Close(); err != nil {
|
||||
h.log.Debugf("error closing websocket: %v", err)
|
||||
}
|
||||
}()
|
||||
|
||||
player.Play()
|
||||
defer player.Close()
|
||||
|
||||
headerBuf := make([]byte, 11)
|
||||
headerBuf[0] = messageTypePTY
|
||||
|
||||
writePTY := func(b []byte, delay uint64) error {
|
||||
writer, err := ws.NextWriter(websocket.BinaryMessage)
|
||||
if err != nil {
|
||||
return trace.Wrap(err, "getting websocket writer")
|
||||
}
|
||||
msgLen := uint16(len(b) + 8)
|
||||
binary.BigEndian.PutUint16(headerBuf[1:], msgLen)
|
||||
binary.BigEndian.PutUint64(headerBuf[3:], delay)
|
||||
if _, err := writer.Write(headerBuf); err != nil {
|
||||
return trace.Wrap(err, "writing message header")
|
||||
}
|
||||
|
||||
// TODO(zmb3): consider optimizing this by bufering for very large sessions
|
||||
// (wait up to N ms to batch events into a single websocket write).
|
||||
if _, err := writer.Write(b); err != nil {
|
||||
return trace.Wrap(err, "writing PTY data")
|
||||
}
|
||||
|
||||
if err := writer.Close(); err != nil {
|
||||
return trace.Wrap(err, "closing websocket writer")
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
writeSize := func(size string) error {
|
||||
ts, err := session.UnmarshalTerminalParams(size)
|
||||
if err != nil {
|
||||
h.log.Debugf("Ignoring invalid terminal size %q", size)
|
||||
return nil // don't abort playback due to a bad event
|
||||
}
|
||||
|
||||
msg := make([]byte, 7)
|
||||
msg[0] = messageTypeResize
|
||||
binary.BigEndian.PutUint16(msg[1:], 4)
|
||||
binary.BigEndian.PutUint16(msg[3:], uint16(ts.W))
|
||||
binary.BigEndian.PutUint16(msg[5:], uint16(ts.H))
|
||||
|
||||
return trace.Wrap(ws.WriteMessage(websocket.BinaryMessage, msg))
|
||||
}
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case evt, ok := <-player.C():
|
||||
if !ok {
|
||||
// send any playback errors to the browser
|
||||
if err := writeError(ws, player.Err()); err != nil {
|
||||
h.log.Warnf("failed to send error message to browser: %v", err)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
switch evt := evt.(type) {
|
||||
case *events.SessionStart:
|
||||
if err := writeSize(evt.TerminalSize); err != nil {
|
||||
h.log.Debugf("Failed to write resize: %v", err)
|
||||
return
|
||||
}
|
||||
|
||||
case *events.SessionPrint:
|
||||
if err := writePTY(evt.Data, uint64(evt.DelayMilliseconds)); err != nil {
|
||||
h.log.Debugf("Failed to send PTY data: %v", err)
|
||||
return
|
||||
}
|
||||
|
||||
case *events.SessionEnd:
|
||||
// send empty PTY data - this will ensure that any dead time
|
||||
// at the end of the recording is processed and the allow
|
||||
// the progress bar to go to 100%
|
||||
if err := writePTY(nil, uint64(evt.EndTime.Sub(evt.StartTime)/time.Millisecond)); err != nil {
|
||||
h.log.Debugf("Failed to send session end data: %v", err)
|
||||
return
|
||||
}
|
||||
|
||||
case *events.Resize:
|
||||
if err := writeSize(evt.TerminalSize); err != nil {
|
||||
h.log.Debugf("Failed to write resize: %v", err)
|
||||
return
|
||||
}
|
||||
|
||||
default:
|
||||
h.log.Debugf("unexpected event type %T", evt)
|
||||
}
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
<-ctx.Done()
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
func writeError(ws *websocket.Conn, err error) error {
|
||||
if err == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
b := new(bytes.Buffer)
|
||||
b.WriteByte(messageTypeError)
|
||||
|
||||
msg := trace.UserMessage(err)
|
||||
l := 1 /* severity */ + 2 /* msg length */ + len(msg)
|
||||
binary.Write(b, binary.BigEndian, uint16(l))
|
||||
b.WriteByte(severityError)
|
||||
binary.Write(b, binary.BigEndian, uint16(len(msg)))
|
||||
b.WriteString(msg)
|
||||
|
||||
return trace.Wrap(ws.WriteMessage(websocket.BinaryMessage, b.Bytes()))
|
||||
}
|
||||
|
||||
type play interface {
|
||||
Play() error
|
||||
Pause() error
|
||||
SetPos(time.Duration) error
|
||||
}
|
||||
|
||||
// handlePlaybackAction processes a playback message
|
||||
// received from the browser
|
||||
func handlePlaybackAction(b []byte, p play) error {
|
||||
if len(b) < 3 {
|
||||
return trace.BadParameter("invalid playback message")
|
||||
}
|
||||
|
||||
msgType := b[0]
|
||||
msgLen := binary.BigEndian.Uint16(b[1:])
|
||||
|
||||
if len(b) < int(msgLen+3) {
|
||||
return trace.BadParameter("invalid message length")
|
||||
}
|
||||
|
||||
payload := b[3:]
|
||||
payload = payload[:msgLen]
|
||||
|
||||
switch msgType {
|
||||
case messageTypePlayPause:
|
||||
if len(payload) != 1 {
|
||||
return trace.BadParameter("invalid play/pause command")
|
||||
}
|
||||
switch action := payload[0]; action {
|
||||
case actionPlay:
|
||||
p.Play()
|
||||
case actionPause:
|
||||
p.Pause()
|
||||
default:
|
||||
return trace.BadParameter("invalid play/pause action %v", action)
|
||||
}
|
||||
case messageTypeSeek:
|
||||
if len(payload) != 8 {
|
||||
return trace.BadParameter("invalid seek message")
|
||||
}
|
||||
pos := binary.BigEndian.Uint64(payload)
|
||||
p.SetPos(time.Duration(pos) * time.Millisecond)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
/*
|
||||
|
||||
# Websocket Protocol for TTY Playback:
|
||||
|
||||
During playback, the Teleport proxy sends session data to the browser
|
||||
and the browser sends playback commands (play/pause, seek, etc) to the
|
||||
proxy.
|
||||
|
||||
Each message conforms to the following binary protocol.
|
||||
|
||||
## Message Header
|
||||
|
||||
The message header starts with a 1-byte identifier followed by a 2-byte
|
||||
(big endian) integer containing the number of bytes following the header.
|
||||
This length field does not include the 3-byte header.
|
||||
|
||||
## Messages
|
||||
|
||||
### 1 - PTY data
|
||||
|
||||
This message is used to send recorded PTY data to the browser.
|
||||
|
||||
- Message ID: 1
|
||||
- 8-byte timestamp (milliseconds since session start)
|
||||
- PTY data
|
||||
|
||||
### 2 - Error
|
||||
|
||||
This message is used to indicate that an error has occurred.
|
||||
|
||||
- Message ID: 2
|
||||
- 1 byte severity (1=error)
|
||||
- 2-byte error message length
|
||||
- variable length error message (UTF-8 text)
|
||||
|
||||
### 3 - Play/Pause
|
||||
|
||||
This message is sent from the browser to the server to pause
|
||||
or resume playback.
|
||||
|
||||
- Message ID: 3
|
||||
- 1-byte code (0=play, 1=pause)
|
||||
|
||||
### 4 - Seek
|
||||
|
||||
This message is used to seek to a new position in the recording.
|
||||
|
||||
- Message ID: 4
|
||||
- 8-byte timestamp (milliseconds since session start)
|
||||
|
||||
### 5 - Resize
|
||||
|
||||
This message is used to indicate that the termina was resized.
|
||||
|
||||
- Message ID: 5
|
||||
- 2-byte width
|
||||
- 2-byte height
|
||||
|
||||
*/
|
||||
@@ -0,0 +1,37 @@
|
||||
/*
|
||||
* Teleport
|
||||
* Copyright (C) 2023 Gravitational, Inc.
|
||||
*
|
||||
* This program is free software: you can redistribute it and/or modify
|
||||
* it under the terms of the GNU Affero General Public License as published by
|
||||
* the Free Software Foundation, either version 3 of the License, or
|
||||
* (at your option) any later version.
|
||||
*
|
||||
* This program is distributed in the hope that it will be useful,
|
||||
* but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
* GNU Affero General Public License for more details.
|
||||
*
|
||||
* You should have received a copy of the GNU Affero General Public License
|
||||
* along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
*/
|
||||
|
||||
package web
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func FuzzHandlePlaybackAction(f *testing.F) {
|
||||
player := nopPlayer{}
|
||||
f.Fuzz(func(t *testing.T, b []byte) {
|
||||
handlePlaybackAction(b, player)
|
||||
})
|
||||
}
|
||||
|
||||
type nopPlayer struct{}
|
||||
|
||||
func (nopPlayer) SetPos(time.Duration) error { return nil }
|
||||
func (nopPlayer) Play() error { return nil }
|
||||
func (nopPlayer) Pause() error { return nil }
|
||||
@@ -25,11 +25,12 @@
|
||||
"xterm-addon-web-links": "^0.8.0"
|
||||
},
|
||||
"devDependencies": {
|
||||
"babel-plugin-transform-import-meta": "^2.2.0",
|
||||
"babel-plugin-transform-vite-meta-env": "^1.0.3",
|
||||
"@gravitational/build": "^1.0.0",
|
||||
"@types/wicg-file-system-access": "^2020.9.5",
|
||||
"babel-plugin-transform-import-meta": "^2.2.0",
|
||||
"babel-plugin-transform-vite-meta-env": "^1.0.3",
|
||||
"jest-canvas-mock": "^2.3.1",
|
||||
"jest-websocket-mock": "^2.5.0",
|
||||
"ts-loader": "^9.4.2"
|
||||
}
|
||||
}
|
||||
|
||||
@@ -101,6 +101,8 @@ export async function resolveServerMessage(
|
||||
}
|
||||
}
|
||||
|
||||
// TODO(zmb3): check with Ryan about replacing this with streaming
|
||||
|
||||
export async function getSessionEvents(sessionUrl: string): Promise<{
|
||||
events: SessionEvent[] | null;
|
||||
}> {
|
||||
|
||||
@@ -36,6 +36,8 @@ import { DesktopPlayer } from './DesktopPlayer';
|
||||
import SshPlayer from './SshPlayer';
|
||||
import Tabs, { TabItem } from './PlayerTabs';
|
||||
|
||||
const validRecordingTypes = ['ssh', 'k8s', 'desktop'];
|
||||
|
||||
export default function Player() {
|
||||
const { sid, clusterId } = useParams<UrlPlayerParams>();
|
||||
const { search } = useLocation();
|
||||
@@ -46,10 +48,7 @@ export default function Player() {
|
||||
) as RecordingType;
|
||||
const durationMs = Number(getUrlParameter('durationMs', search));
|
||||
|
||||
const validRecordingType =
|
||||
recordingType === 'ssh' ||
|
||||
recordingType === 'k8s' ||
|
||||
recordingType === 'desktop';
|
||||
const validRecordingType = validRecordingTypes.includes(recordingType);
|
||||
const validDurationMs = Number.isInteger(durationMs) && durationMs > 0;
|
||||
|
||||
document.title = `${clusterId} • Play ${sid}`;
|
||||
@@ -64,14 +63,14 @@ export default function Player() {
|
||||
<Box textAlign="center" mx={10} mt={5}>
|
||||
<Danger mb={0}>
|
||||
Invalid query parameter recordingType: {recordingType}, should be
|
||||
'ssh' or 'desktop'
|
||||
one of {validRecordingTypes.join(', ')}.
|
||||
</Danger>
|
||||
</Box>
|
||||
</StyledPlayer>
|
||||
);
|
||||
}
|
||||
|
||||
if (recordingType === 'desktop' && !validDurationMs) {
|
||||
if (!validDurationMs) {
|
||||
return (
|
||||
<StyledPlayer>
|
||||
<Box textAlign="center" mx={10} mt={5}>
|
||||
@@ -106,7 +105,7 @@ export default function Player() {
|
||||
durationMs={durationMs}
|
||||
/>
|
||||
) : (
|
||||
<SshPlayer sid={sid} clusterId={clusterId} />
|
||||
<SshPlayer sid={sid} clusterId={clusterId} durationMs={durationMs} />
|
||||
)}
|
||||
</Flex>
|
||||
</StyledPlayer>
|
||||
|
||||
@@ -26,7 +26,7 @@ export default function ProgressBar(props: ProgressBarProps) {
|
||||
const Icon = props.isPlaying ? Icons.CirclePause : Icons.CirclePlay;
|
||||
return (
|
||||
<StyledProgessBar style={props.style} id={props.id}>
|
||||
<ActionButton onClick={props.toggle}>
|
||||
<ActionButton onClick={props.toggle} disabled={props.disabled}>
|
||||
<Icon />
|
||||
</ActionButton>
|
||||
<PlaySpeedSelector onChange={props.onPlaySpeedChange} />
|
||||
@@ -36,7 +36,9 @@ export default function ProgressBar(props: ProgressBarProps) {
|
||||
min={props.min}
|
||||
max={props.max}
|
||||
value={props.current}
|
||||
onChange={props.move}
|
||||
disabled={props.disabled}
|
||||
onBeforeChange={props.onStartMove}
|
||||
onAfterChange={props.move}
|
||||
defaultValue={1}
|
||||
withBars
|
||||
className="grv-slider"
|
||||
@@ -67,13 +69,15 @@ function Restart(props: { onRestart?: () => void }) {
|
||||
export type ProgressBarProps = {
|
||||
max: number;
|
||||
min: number;
|
||||
time: any;
|
||||
time: string;
|
||||
isPlaying: boolean;
|
||||
disabled?: boolean;
|
||||
current: number;
|
||||
move: (value: any) => void;
|
||||
toggle: () => void;
|
||||
style?: React.CSSProperties;
|
||||
id?: string;
|
||||
onStartMove?: () => void;
|
||||
onPlaySpeedChange?: (newSpeed: number) => void;
|
||||
onRestart?: () => void;
|
||||
};
|
||||
@@ -140,7 +144,13 @@ const ActionButton = styled.button`
|
||||
color: ${props => props.theme.colors.text.main};
|
||||
}
|
||||
|
||||
&:hover {
|
||||
&:disabled {
|
||||
.icon {
|
||||
color: ${props => props.theme.colors.text.disabled};
|
||||
}
|
||||
}
|
||||
|
||||
&:hover:enabled {
|
||||
opacity: 1;
|
||||
|
||||
.icon {
|
||||
|
||||
@@ -1,87 +0,0 @@
|
||||
/**
|
||||
* Teleport
|
||||
* Copyright (C) 2023 Gravitational, Inc.
|
||||
*
|
||||
* This program is free software: you can redistribute it and/or modify
|
||||
* it under the terms of the GNU Affero General Public License as published by
|
||||
* the Free Software Foundation, either version 3 of the License, or
|
||||
* (at your option) any later version.
|
||||
*
|
||||
* This program is distributed in the hope that it will be useful,
|
||||
* but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
* GNU Affero General Public License for more details.
|
||||
*
|
||||
* You should have received a copy of the GNU Affero General Public License
|
||||
* along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
*/
|
||||
|
||||
import React from 'react';
|
||||
import { throttle } from 'shared/utils/highbar';
|
||||
|
||||
import TtyPlayer from 'teleport/lib/term/ttyPlayer';
|
||||
|
||||
import ProgressBar from './ProgressBar';
|
||||
|
||||
export default function ProgressBarTty(props: { tty: TtyPlayer }) {
|
||||
const state = useTtyProgress(props.tty);
|
||||
return <ProgressBar {...state} />;
|
||||
}
|
||||
|
||||
export function useTtyProgress(tty: TtyPlayer) {
|
||||
const [state, setState] = React.useState(() => {
|
||||
return makeTtyProgress(tty);
|
||||
});
|
||||
|
||||
React.useEffect(() => {
|
||||
const throttledOnChange = throttle(
|
||||
onChange,
|
||||
// some magic numbers to reduce number of re-renders when
|
||||
// session is too long and "eventful"
|
||||
Math.max(Math.min(tty.duration * 0.025, 500), 20)
|
||||
);
|
||||
|
||||
function onChange() {
|
||||
// recalculate progress state
|
||||
const ttyProgres = makeTtyProgress(tty);
|
||||
setState(ttyProgres);
|
||||
}
|
||||
|
||||
function cleanup() {
|
||||
throttledOnChange.cancel();
|
||||
tty.stop();
|
||||
tty.removeAllListeners();
|
||||
}
|
||||
|
||||
tty.on('change', throttledOnChange);
|
||||
|
||||
return cleanup;
|
||||
}, [tty]);
|
||||
|
||||
return state;
|
||||
}
|
||||
|
||||
function makeTtyProgress(tty: TtyPlayer) {
|
||||
function toggle() {
|
||||
if (tty.isPlaying()) {
|
||||
tty.stop();
|
||||
} else {
|
||||
tty.play();
|
||||
}
|
||||
}
|
||||
|
||||
function move(value) {
|
||||
tty.move(value);
|
||||
}
|
||||
|
||||
return {
|
||||
max: tty.duration,
|
||||
min: 1,
|
||||
time: tty.getCurrentTime(),
|
||||
isLoading: tty.isLoading(),
|
||||
isPlaying: tty.isPlaying(),
|
||||
current: tty.current,
|
||||
move,
|
||||
toggle,
|
||||
};
|
||||
}
|
||||
@@ -1,21 +1,3 @@
|
||||
/*
|
||||
* Teleport
|
||||
* Copyright (C) 2023 Gravitational, Inc.
|
||||
*
|
||||
* This program is free software: you can redistribute it and/or modify
|
||||
* it under the terms of the GNU Affero General Public License as published by
|
||||
* the Free Software Foundation, either version 3 of the License, or
|
||||
* (at your option) any later version.
|
||||
*
|
||||
* This program is distributed in the hope that it will be useful,
|
||||
* but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
* GNU Affero General Public License for more details.
|
||||
*
|
||||
* You should have received a copy of the GNU Affero General Public License
|
||||
* along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
*/
|
||||
|
||||
/*
|
||||
The MIT License (MIT)
|
||||
|
||||
|
||||
@@ -17,8 +17,7 @@
|
||||
*/
|
||||
|
||||
import ProgressBar from './ProgressBar';
|
||||
import ProgressBarTty from './ProgressBarTty';
|
||||
import { ProgressBarDesktop } from './ProgressBarDesktop';
|
||||
|
||||
export default ProgressBar;
|
||||
export { ProgressBarTty, ProgressBarDesktop };
|
||||
export { ProgressBarDesktop };
|
||||
|
||||
@@ -18,24 +18,27 @@
|
||||
|
||||
import React from 'react';
|
||||
import styled from 'styled-components';
|
||||
import { Indicator, Flex, Box } from 'design';
|
||||
import { Danger } from 'design/Alert';
|
||||
import { Indicator, Flex, Text, Box } from 'design';
|
||||
|
||||
import cfg from 'teleport/config';
|
||||
import TtyPlayer, {
|
||||
StatusEnum,
|
||||
StatusEnum as TtyStatusEnum,
|
||||
} from 'teleport/lib/term/ttyPlayer';
|
||||
import EventProvider from 'teleport/lib/term/ttyPlayerEventProvider';
|
||||
import { getAccessToken, getHostName } from 'teleport/services/api';
|
||||
|
||||
import { ProgressBarTty } from './ProgressBar';
|
||||
import ProgressBar from './ProgressBar';
|
||||
import Xterm from './Xterm';
|
||||
|
||||
export default function Player({ sid, clusterId }) {
|
||||
const { tty } = useSshPlayer(clusterId, sid);
|
||||
const { statusText, status } = tty;
|
||||
const eventCount = tty.getEventCount();
|
||||
const isError = status === TtyStatusEnum.ERROR;
|
||||
const isLoading = status === TtyStatusEnum.LOADING;
|
||||
export default function Player({ sid, clusterId, durationMs }) {
|
||||
const { tty, playerStatus, statusText, time } = useStreamingSshPlayer(
|
||||
clusterId,
|
||||
sid
|
||||
);
|
||||
const isError = playerStatus === TtyStatusEnum.ERROR;
|
||||
const isLoading = playerStatus === TtyStatusEnum.LOADING;
|
||||
const isPlaying = playerStatus === TtyStatusEnum.PLAYING;
|
||||
|
||||
if (isError) {
|
||||
return (
|
||||
@@ -53,22 +56,31 @@ export default function Player({ sid, clusterId }) {
|
||||
);
|
||||
}
|
||||
|
||||
if (!isLoading && eventCount === 0) {
|
||||
return (
|
||||
<StatusBox>
|
||||
<Text typography="h4">
|
||||
Recording for this session is not available.
|
||||
</Text>
|
||||
</StatusBox>
|
||||
);
|
||||
}
|
||||
|
||||
return (
|
||||
<StyledPlayer>
|
||||
<Flex flex="1" flexDirection="column" overflow="auto">
|
||||
<Xterm tty={tty} />
|
||||
</Flex>
|
||||
{eventCount > 0 && <ProgressBarTty tty={tty} />}
|
||||
<ProgressBar
|
||||
min={0}
|
||||
max={durationMs}
|
||||
current={time}
|
||||
disabled={
|
||||
playerStatus === TtyStatusEnum.ERROR ||
|
||||
playerStatus === TtyStatusEnum.COMPLETE
|
||||
}
|
||||
isPlaying={isPlaying}
|
||||
time={formatDisplayTime(time)}
|
||||
onRestart={window.location.reload}
|
||||
onStartMove={tty.suspendTimeUpdates}
|
||||
move={pos => {
|
||||
tty.move(pos);
|
||||
tty.resumeTimeUpdates();
|
||||
}}
|
||||
toggle={() => {
|
||||
isPlaying ? tty.stop() : tty.play();
|
||||
}}
|
||||
/>
|
||||
</StyledPlayer>
|
||||
);
|
||||
}
|
||||
@@ -87,35 +99,45 @@ const StyledPlayer = styled.div`
|
||||
justify-content: space-between;
|
||||
`;
|
||||
|
||||
function useSshPlayer(clusterId: string, sid: string) {
|
||||
const tty = React.useMemo(() => {
|
||||
const prefixUrl = cfg.getSshPlaybackPrefixUrl({ clusterId, sid });
|
||||
return new TtyPlayer(new EventProvider({ url: prefixUrl }));
|
||||
}, [sid, clusterId]);
|
||||
function useStreamingSshPlayer(clusterId: string, sid: string) {
|
||||
const [playerStatus, setPlayerStatus] = React.useState(StatusEnum.LOADING);
|
||||
const [statusText, setStatusText] = React.useState('');
|
||||
const [time, setTime] = React.useState(0);
|
||||
|
||||
// to trigger re-render when tty state changes
|
||||
const [, rerender] = React.useState(tty.status);
|
||||
const tty = React.useMemo(() => {
|
||||
const url = cfg.api.ttyPlaybackWsAddr
|
||||
.replace(':fqdn', getHostName())
|
||||
.replace(':clusterId', clusterId)
|
||||
.replace(':sid', sid)
|
||||
.replace(':token', getAccessToken());
|
||||
return new TtyPlayer({ url, setPlayerStatus, setStatusText, setTime });
|
||||
}, [clusterId, sid, setPlayerStatus, setStatusText, setTime]);
|
||||
|
||||
React.useEffect(() => {
|
||||
function onChange() {
|
||||
// trigger rerender when status changes
|
||||
rerender(tty.status);
|
||||
}
|
||||
tty.connect();
|
||||
tty.play();
|
||||
|
||||
function cleanup() {
|
||||
return () => {
|
||||
tty.stop();
|
||||
tty.removeAllListeners();
|
||||
}
|
||||
|
||||
tty.on('change', onChange);
|
||||
tty.connect().then(() => {
|
||||
tty.play();
|
||||
});
|
||||
|
||||
return cleanup;
|
||||
};
|
||||
}, [tty]);
|
||||
|
||||
return {
|
||||
tty,
|
||||
};
|
||||
return { tty, playerStatus, statusText, time };
|
||||
}
|
||||
|
||||
function formatDisplayTime(ms: number) {
|
||||
if (ms <= 0) {
|
||||
return '00:00';
|
||||
}
|
||||
|
||||
const totalSec = Math.floor(ms / 1000);
|
||||
const totalDays = (totalSec % 31536000) % 86400;
|
||||
const h = Math.floor(totalDays / 3600);
|
||||
const m = Math.floor((totalDays % 3600) / 60);
|
||||
const s = (totalDays % 3600) % 60;
|
||||
|
||||
return `${h > 0 ? h + ':' : ''}${m.toString().padStart(2, '0')}:${s
|
||||
.toString()
|
||||
.padStart(2, '0')}`;
|
||||
}
|
||||
|
||||
@@ -134,7 +134,7 @@ const renderPlayCell = (
|
||||
{ clusterId, sid },
|
||||
{
|
||||
recordingType,
|
||||
durationMs: recordingType === 'desktop' ? duration : undefined,
|
||||
durationMs: duration,
|
||||
}
|
||||
);
|
||||
return (
|
||||
|
||||
@@ -915,7 +915,7 @@ exports[`rendering of Session Recordings 1`] = `
|
||||
>
|
||||
<a
|
||||
class="c28"
|
||||
href="/web/cluster/localhost/session/8efccedd-9633-473f-bfb3-fcc07e2af345?recordingType=k8s"
|
||||
href="/web/cluster/localhost/session/8efccedd-9633-473f-bfb3-fcc07e2af345?recordingType=k8s&durationMs=9477"
|
||||
kind="primary"
|
||||
target="_blank"
|
||||
width="80px"
|
||||
@@ -1051,7 +1051,7 @@ exports[`rendering of Session Recordings 1`] = `
|
||||
>
|
||||
<a
|
||||
class="c28"
|
||||
href="/web/cluster/localhost/session/426485-6491-11e9-80a1-427cfde50f5a?recordingType=ssh"
|
||||
href="/web/cluster/localhost/session/426485-6491-11e9-80a1-427cfde50f5a?recordingType=ssh&durationMs=1161287"
|
||||
kind="primary"
|
||||
target="_blank"
|
||||
width="80px"
|
||||
@@ -1101,7 +1101,7 @@ exports[`rendering of Session Recordings 1`] = `
|
||||
>
|
||||
<a
|
||||
class="c28"
|
||||
href="/web/cluster/localhost/session/377875-6491-11e9-80a1-427cfde50f5a?recordingType=ssh"
|
||||
href="/web/cluster/localhost/session/377875-6491-11e9-80a1-427cfde50f5a?recordingType=ssh&durationMs=21287"
|
||||
kind="primary"
|
||||
target="_blank"
|
||||
width="80px"
|
||||
|
||||
@@ -200,7 +200,11 @@ const cfg = {
|
||||
desktopIsActive: '/v1/webapi/sites/:clusterId/desktops/:desktopName/active',
|
||||
ttyWsAddr:
|
||||
'wss://:fqdn/v1/webapi/sites/:clusterId/connect?access_token=:token¶ms=:params&traceparent=:traceparent',
|
||||
ttyPlaybackWsAddr:
|
||||
'wss://:fqdn/v1/webapi/sites/:clusterId/ttyplayback/:sid?access_token=:token', // TODO(zmb3): get token out of URL
|
||||
activeAndPendingSessionsPath: '/v1/webapi/sites/:clusterId/sessions',
|
||||
|
||||
// TODO(zmb3): remove this for v15
|
||||
sshPlaybackPrefix: '/v1/webapi/sites/:clusterId/sessions/:sid', // prefix because this is eventually concatenated with "/stream" or "/events"
|
||||
kubernetesPath:
|
||||
'/v1/webapi/sites/:clusterId/kubernetes?searchAsRoles=:searchAsRoles?&limit=:limit?&startKey=:startKey?&query=:query?&search=:search?&sort=:sort?',
|
||||
@@ -594,6 +598,7 @@ const cfg = {
|
||||
},
|
||||
|
||||
getSshPlaybackPrefixUrl({ clusterId, sid }: UrlParams) {
|
||||
// TODO(zmb3): remove
|
||||
return generatePath(cfg.api.sshPlaybackPrefix, { clusterId, sid });
|
||||
},
|
||||
|
||||
@@ -1023,7 +1028,7 @@ export interface UrlPlayerParams {
|
||||
|
||||
export interface UrlPlayerSearch {
|
||||
recordingType: RecordingType;
|
||||
durationMs?: number; // this is only necessary for recordingType == desktop
|
||||
durationMs?: number;
|
||||
}
|
||||
|
||||
// /web/cluster/:clusterId/desktops/:desktopName/:username
|
||||
|
||||
File diff suppressed because one or more lines are too long
@@ -158,7 +158,11 @@ export default class TtyTerminal {
|
||||
_processData(data) {
|
||||
try {
|
||||
this.tty.pauseFlow();
|
||||
this.term.write(data, () => this.tty.resumeFlow());
|
||||
|
||||
// during a live session, data is emitted as a string.
|
||||
// during playback, data from the websocket comes over as a DataView
|
||||
const d: any = typeof data === 'string' ? data : new Uint8Array(data);
|
||||
this.term.write(d, () => this.tty.resumeFlow());
|
||||
} catch (err) {
|
||||
logger.error('xterm.write', data, err);
|
||||
// recover xtermjs by resetting it
|
||||
|
||||
@@ -181,6 +181,7 @@ class Tty extends EventEmitterWebAuthnSender {
|
||||
try {
|
||||
const uintArray = new Uint8Array(ev.data);
|
||||
const msg = this._proto.decode(uintArray);
|
||||
|
||||
switch (msg.type) {
|
||||
case MessageTypeEnum.WEBAUTHN_CHALLENGE:
|
||||
this.emit(TermEvent.WEBAUTHN_CHALLENGE, msg.payload);
|
||||
|
||||
@@ -16,7 +16,11 @@
|
||||
* along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
*/
|
||||
|
||||
export default class AddressResolver {
|
||||
/**
|
||||
* LiveSessionAddressResolver is an address resolver that computes
|
||||
* a URL to start a new web-based SSH session.
|
||||
*/
|
||||
export default class LiveSessionAddressResolver {
|
||||
_cfg = {
|
||||
ttyUrl: null,
|
||||
ttyParams: {},
|
||||
|
||||
@@ -16,263 +16,235 @@
|
||||
* along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
*/
|
||||
|
||||
import BufferModule from 'buffer/';
|
||||
import { throttle } from 'shared/utils/highbar';
|
||||
import Logger from 'shared/libs/logger';
|
||||
|
||||
import Tty from './tty';
|
||||
import { TermEvent } from './enums';
|
||||
import { onlyPrintEvents } from './ttyPlayerEventProvider';
|
||||
import { TermEvent, WebsocketCloseCode } from './enums';
|
||||
|
||||
const logger = Logger.create('TtyPlayer');
|
||||
const STREAM_START_INDEX = 0;
|
||||
const PLAY_SPEED = 10;
|
||||
|
||||
export const Buffer = BufferModule.Buffer;
|
||||
export const StatusEnum = {
|
||||
PLAYING: 'PLAYING',
|
||||
ERROR: 'ERROR',
|
||||
PAUSED: 'PAUSED',
|
||||
LOADING: 'LOADING',
|
||||
COMPLETE: 'COMPLETE',
|
||||
};
|
||||
|
||||
const messageTypePty = 1;
|
||||
const messageTypeError = 2;
|
||||
const messageTypePlayPause = 3;
|
||||
const messageTypeSeek = 4;
|
||||
const messageTypeResize = 5;
|
||||
|
||||
const actionPlay = 0;
|
||||
const actionPause = 1;
|
||||
|
||||
// we update the time every time we receive data, or
|
||||
// at this interval (which ensures that the progress
|
||||
// bar updates even when we aren't receiving data)
|
||||
const PROGRESS_UPDATE_INTERVAL_MS = 50;
|
||||
|
||||
export default class TtyPlayer extends Tty {
|
||||
constructor(eventProvider) {
|
||||
constructor({ url, setPlayerStatus, setStatusText, setTime }) {
|
||||
super({});
|
||||
this.currentEventIndex = 0;
|
||||
this.current = 0;
|
||||
this.duration = 0;
|
||||
this.status = StatusEnum.LOADING;
|
||||
this.statusText = '';
|
||||
|
||||
this._posToEventIndexMap = [];
|
||||
this._eventProvider = eventProvider;
|
||||
this._url = url;
|
||||
this._setPlayerStatus = setPlayerStatus;
|
||||
this._setStatusText = setStatusText;
|
||||
|
||||
// _chunkQueue is a list of data chunks waiting to be rendered by the term.
|
||||
this._chunkQueue = [];
|
||||
// _writeInFlight prevents sending more data to xterm while a prior render has not finished yet.
|
||||
this._writeInFlight = false;
|
||||
this._paused = false;
|
||||
this._lastPlayedTimestamp = 0;
|
||||
|
||||
this._sendTimeUpdates = true;
|
||||
this._setTime = throttle(t => setTime(t), PROGRESS_UPDATE_INTERVAL_MS);
|
||||
this._lastUpdate = 0;
|
||||
this._timeout = null;
|
||||
}
|
||||
|
||||
// Override the base class connection, which uses the envelope-based
|
||||
// websocket protocol (this protocol doesn't support sending timing data).
|
||||
connect() {
|
||||
this._setPlayerStatus(StatusEnum.LOADING);
|
||||
|
||||
this.webSocket = new WebSocket(this._url);
|
||||
this.webSocket.binaryType = 'arraybuffer';
|
||||
this.webSocket.onopen = () => this.emit('open');
|
||||
this.webSocket.onmessage = m => this.onMessage(m);
|
||||
this.webSocket.onclose = e => {
|
||||
logger.debug('websocket closed', e);
|
||||
this.cancelTimeUpdate();
|
||||
|
||||
this.webSocket.close();
|
||||
this.webSocket.onopen = null;
|
||||
this.webSocket.onclose = null;
|
||||
this.webSocket.onmessage = null;
|
||||
this.webSocket = null;
|
||||
|
||||
this.emit(TermEvent.CONN_CLOSE, e);
|
||||
this._setPlayerStatus(StatusEnum.COMPLETE);
|
||||
};
|
||||
}
|
||||
|
||||
suspendTimeUpdates() {
|
||||
this._sendTimeUpdates = false;
|
||||
}
|
||||
|
||||
resumeTimeUpdates() {
|
||||
this._sendTimeUpdates = true;
|
||||
}
|
||||
|
||||
setTime(t) {
|
||||
// time updates are suspended when a user is dragging the slider to
|
||||
// a new position (it's very disruptive if we're updating the slider
|
||||
// position every few milliseconds while the user is trying to
|
||||
// reposition it)
|
||||
if (this._sendTimeUpdates) {
|
||||
this._setTime(t);
|
||||
}
|
||||
}
|
||||
|
||||
disconnect(closeCode = WebsocketCloseCode.NORMAL) {
|
||||
this.cancelTimeUpdate();
|
||||
if (this.webSocket !== null) {
|
||||
this.webSocket.close(closeCode);
|
||||
}
|
||||
}
|
||||
|
||||
scheduleNextUpdate(current) {
|
||||
this._timeout = setTimeout(() => {
|
||||
const delta = Date.now() - this._lastUpdate;
|
||||
const next = current + delta;
|
||||
this.setTime(next);
|
||||
this._lastUpdate = Date.now();
|
||||
|
||||
this.scheduleNextUpdate(next);
|
||||
}, PROGRESS_UPDATE_INTERVAL_MS);
|
||||
}
|
||||
|
||||
cancelTimeUpdate() {
|
||||
if (this._timeout != null) {
|
||||
clearTimeout(this._timeout);
|
||||
this._timeout = null;
|
||||
}
|
||||
}
|
||||
|
||||
onMessage(m) {
|
||||
try {
|
||||
const dv = new DataView(m.data);
|
||||
const typ = dv.getUint8(0);
|
||||
const len = dv.getUint16(1);
|
||||
|
||||
// see lib/web/tty_playback.go for details on this protocol
|
||||
switch (typ) {
|
||||
case messageTypePty:
|
||||
this.cancelTimeUpdate();
|
||||
|
||||
const delay = Number(dv.getBigUint64(3));
|
||||
const data = dv.buffer.slice(
|
||||
dv.byteOffset + 11,
|
||||
dv.byteOffset + 11 + len
|
||||
);
|
||||
|
||||
this.emit(TermEvent.DATA, data);
|
||||
this._lastPlayedTimestamp = delay;
|
||||
|
||||
this._lastUpdate = Date.now();
|
||||
this.setTime(delay);
|
||||
|
||||
// schedule the next time update (in case this
|
||||
// part of the recording is dead time)
|
||||
// TODO(zmb3): implement this for desktops too
|
||||
if (!this._paused) {
|
||||
this.scheduleNextUpdate(delay);
|
||||
}
|
||||
break;
|
||||
|
||||
case messageTypeError:
|
||||
// ignore the severity byte at index 3 (we display all errors identically)
|
||||
const msgLen = dv.getUint16(4);
|
||||
const msg = new TextDecoder().decode(
|
||||
dv.buffer.slice(dv.byteOffset + 6, dv.byteOffset + 6 + msgLen)
|
||||
);
|
||||
this._setStatusText(msg);
|
||||
this._setPlayerStatus(StatusEnum.ERROR);
|
||||
this.disconnect();
|
||||
return;
|
||||
|
||||
case messageTypeResize:
|
||||
const w = dv.getUint16(3);
|
||||
const h = dv.getUint16(5);
|
||||
this.emit(TermEvent.RESIZE, { w, h });
|
||||
return;
|
||||
|
||||
default:
|
||||
logger.warn('unexpected message type', typ);
|
||||
return;
|
||||
}
|
||||
} catch (err) {
|
||||
logger.error('failed to parse incoming message', err);
|
||||
}
|
||||
}
|
||||
|
||||
// override
|
||||
send() {}
|
||||
|
||||
// override
|
||||
connect() {
|
||||
this.status = StatusEnum.LOADING;
|
||||
this._change();
|
||||
return this._eventProvider
|
||||
.init()
|
||||
.then(() => {
|
||||
this._init();
|
||||
this.status = StatusEnum.PAUSED;
|
||||
})
|
||||
.catch(err => {
|
||||
logger.error('unable to init event provider', err);
|
||||
this._handleError(err);
|
||||
})
|
||||
.finally(this._change.bind(this));
|
||||
}
|
||||
|
||||
pauseFlow() {
|
||||
this._writeInFlight = true;
|
||||
}
|
||||
|
||||
resumeFlow() {
|
||||
this._writeInFlight = false;
|
||||
this._chunkDequeue();
|
||||
}
|
||||
pauseFlow() {}
|
||||
resumeFlow() {}
|
||||
|
||||
move(newPos) {
|
||||
if (!this.isReady()) {
|
||||
return;
|
||||
}
|
||||
|
||||
if (newPos === undefined) {
|
||||
newPos = this.current + 1;
|
||||
}
|
||||
|
||||
if (newPos < 0) {
|
||||
newPos = 0;
|
||||
}
|
||||
|
||||
if (newPos > this.duration) {
|
||||
this.stop();
|
||||
}
|
||||
|
||||
const newEventIndex = this._getEventIndex(newPos) + 1;
|
||||
|
||||
if (newEventIndex === this.currentEventIndex) {
|
||||
this.current = newPos;
|
||||
this._change();
|
||||
return;
|
||||
}
|
||||
|
||||
const isRewind = this.currentEventIndex > newEventIndex;
|
||||
this.cancelTimeUpdate();
|
||||
|
||||
try {
|
||||
// we cannot playback the content within terminal so instead:
|
||||
// 1. tell terminal to reset.
|
||||
// 2. tell terminal to render 1 huge chunk that has everything up to current
|
||||
// location.
|
||||
if (isRewind) {
|
||||
this._chunkQueue = [];
|
||||
this.emit(TermEvent.RESET);
|
||||
}
|
||||
const buffer = new ArrayBuffer(11);
|
||||
const dv = new DataView(buffer);
|
||||
dv.setUint8(0, messageTypeSeek);
|
||||
dv.setUint16(1, 8 /* length */);
|
||||
dv.setBigUint64(3, BigInt(newPos));
|
||||
this.webSocket.send(dv);
|
||||
} catch (e) {
|
||||
logger.error('error seeking', e);
|
||||
}
|
||||
|
||||
const from = isRewind ? 0 : this.currentEventIndex;
|
||||
const to = newEventIndex;
|
||||
const events = this._eventProvider.events.slice(from, to);
|
||||
const printEvents = events.filter(onlyPrintEvents);
|
||||
|
||||
this._render(printEvents);
|
||||
this.currentEventIndex = newEventIndex;
|
||||
this.current = newPos;
|
||||
this._change();
|
||||
} catch (err) {
|
||||
logger.error('move', err);
|
||||
this._handleError(err);
|
||||
if (newPos < this._lastPlayedTimestamp) {
|
||||
this.emit(TermEvent.RESET);
|
||||
} else if (this._paused) {
|
||||
// if we're paused, we want the scrubber to "stick" at the new
|
||||
// time until we press play (rather than waiting for us to click
|
||||
// play and start receiving new data)
|
||||
this._setTime(newPos);
|
||||
}
|
||||
}
|
||||
|
||||
stop() {
|
||||
this.status = StatusEnum.PAUSED;
|
||||
this.timer = clearInterval(this.timer);
|
||||
this._change();
|
||||
this._paused = true;
|
||||
this.cancelTimeUpdate();
|
||||
this._setPlayerStatus(StatusEnum.PAUSED);
|
||||
|
||||
const buffer = new ArrayBuffer(4);
|
||||
const dv = new DataView(buffer);
|
||||
dv.setUint8(0, messageTypePlayPause);
|
||||
dv.setUint16(1, 1 /* size */);
|
||||
dv.setUint8(3, actionPause);
|
||||
this.webSocket.send(dv);
|
||||
}
|
||||
|
||||
play() {
|
||||
if (this.status === StatusEnum.PLAYING) {
|
||||
this._paused = false;
|
||||
this._setPlayerStatus(StatusEnum.PLAYING);
|
||||
|
||||
// the very first play call happens before we've even
|
||||
// connected - we only need to send the websocket message
|
||||
// for subsequent calls
|
||||
if (this.webSocket.readyState !== WebSocket.OPEN) {
|
||||
return;
|
||||
}
|
||||
|
||||
this.status = StatusEnum.PLAYING;
|
||||
// start from the beginning if reached the end of the session
|
||||
if (this.current >= this.duration) {
|
||||
this.current = STREAM_START_INDEX;
|
||||
this.emit(TermEvent.RESET);
|
||||
}
|
||||
|
||||
this.timer = setInterval(this.move.bind(this), PLAY_SPEED);
|
||||
this._change();
|
||||
}
|
||||
|
||||
getCurrentTime() {
|
||||
if (this.currentEventIndex) {
|
||||
let { displayTime } =
|
||||
this._eventProvider.events[this.currentEventIndex - 1];
|
||||
return displayTime;
|
||||
} else {
|
||||
return '--:--';
|
||||
}
|
||||
}
|
||||
|
||||
getEventCount() {
|
||||
return this._eventProvider.events.length;
|
||||
}
|
||||
|
||||
isLoading() {
|
||||
return this.status === StatusEnum.LOADING;
|
||||
}
|
||||
|
||||
isPlaying() {
|
||||
return this.status === StatusEnum.PLAYING;
|
||||
}
|
||||
|
||||
isError() {
|
||||
return this.status === StatusEnum.ERROR;
|
||||
}
|
||||
|
||||
isReady() {
|
||||
return (
|
||||
this.status !== StatusEnum.LOADING && this.status !== StatusEnum.ERROR
|
||||
);
|
||||
}
|
||||
|
||||
disconnect() {
|
||||
// do nothing
|
||||
}
|
||||
|
||||
_init() {
|
||||
this.duration = this._eventProvider.getDuration();
|
||||
this._eventProvider.events.forEach(item =>
|
||||
this._posToEventIndexMap.push(item.msNormalized)
|
||||
);
|
||||
}
|
||||
|
||||
_chunkDequeue() {
|
||||
const chunk = this._chunkQueue.shift();
|
||||
if (!chunk) {
|
||||
return;
|
||||
}
|
||||
|
||||
const str = chunk.data.join('');
|
||||
this.emit(TermEvent.RESIZE, { h: chunk.h, w: chunk.w });
|
||||
this.emit(TermEvent.DATA, str);
|
||||
}
|
||||
|
||||
_render(events) {
|
||||
if (!events || events.length === 0) {
|
||||
return;
|
||||
}
|
||||
|
||||
const groups = [
|
||||
{
|
||||
data: [events[0].data],
|
||||
w: events[0].w,
|
||||
h: events[0].h,
|
||||
},
|
||||
];
|
||||
|
||||
let cur = groups[0];
|
||||
|
||||
// group events by screen size and construct 1 chunk of data per group
|
||||
for (let i = 1; i < events.length; i++) {
|
||||
if (cur.w === events[i].w && cur.h === events[i].h) {
|
||||
cur.data.push(events[i].data);
|
||||
} else {
|
||||
cur = {
|
||||
data: [events[i].data],
|
||||
w: events[i].w,
|
||||
h: events[i].h,
|
||||
};
|
||||
|
||||
groups.push(cur);
|
||||
}
|
||||
}
|
||||
|
||||
this._chunkQueue = [...this._chunkQueue, ...groups];
|
||||
if (!this._writeInFlight) {
|
||||
this._chunkDequeue();
|
||||
}
|
||||
}
|
||||
|
||||
_getEventIndex(num) {
|
||||
const arr = this._posToEventIndexMap;
|
||||
var low = 0;
|
||||
var hi = arr.length - 1;
|
||||
|
||||
while (hi - low > 1) {
|
||||
const mid = Math.floor((low + hi) / 2);
|
||||
if (arr[mid] < num) {
|
||||
low = mid;
|
||||
} else {
|
||||
hi = mid;
|
||||
}
|
||||
}
|
||||
|
||||
if (num - arr[low] <= arr[hi] - num) {
|
||||
return low;
|
||||
}
|
||||
|
||||
return hi;
|
||||
}
|
||||
|
||||
_change() {
|
||||
this.emit('change');
|
||||
}
|
||||
|
||||
_handleError(err) {
|
||||
this.status = StatusEnum.ERROR;
|
||||
this.statusText = err.message;
|
||||
const buffer = new ArrayBuffer(4);
|
||||
const dv = new DataView(buffer);
|
||||
dv.setUint8(0, messageTypePlayPause);
|
||||
dv.setUint16(1, 1 /* size */);
|
||||
dv.setUint8(3, actionPlay);
|
||||
this.webSocket.send(dv);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -17,247 +17,127 @@
|
||||
*/
|
||||
|
||||
import '@gravitational/shared/libs/polyfillFinally';
|
||||
import api from 'teleport/services/api';
|
||||
|
||||
import WS from 'jest-websocket-mock';
|
||||
|
||||
import { TermEvent } from 'teleport/lib/term/enums';
|
||||
|
||||
import TtyPlayer, { Buffer } from './ttyPlayer';
|
||||
import EventProvider, { MAX_SIZE } from './ttyPlayerEventProvider';
|
||||
import sample from './fixtures/streamData';
|
||||
|
||||
describe('lib/term/ttyPlayer/eventProvider', () => {
|
||||
afterEach(function () {
|
||||
jest.clearAllMocks();
|
||||
});
|
||||
|
||||
it('should create an instance', () => {
|
||||
const provider = new EventProvider({ url: 'sample.com' });
|
||||
expect(provider.events).toEqual([]);
|
||||
});
|
||||
|
||||
it('should load events and initialize itself', async () => {
|
||||
const provider = new EventProvider({ url: 'sample.com' });
|
||||
|
||||
jest.spyOn(api, 'get').mockImplementation(() => Promise.resolve(sample));
|
||||
jest.spyOn(provider, '_createEvents');
|
||||
jest.spyOn(provider, '_normalizeEventsByTime');
|
||||
jest
|
||||
.spyOn(provider, '_fetchContent')
|
||||
.mockImplementation(() => Promise.resolve());
|
||||
jest.spyOn(provider, '_populatePrintEvents').mockImplementation();
|
||||
|
||||
await provider.init();
|
||||
|
||||
expect(api.get).toHaveBeenCalledWith('sample.com/events');
|
||||
expect(provider._createEvents).toHaveBeenCalledWith(sample.events);
|
||||
expect(provider._normalizeEventsByTime).toHaveBeenCalled();
|
||||
expect(provider._fetchContent).toHaveBeenCalled();
|
||||
expect(provider._populatePrintEvents).toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('should create event objects', () => {
|
||||
const provider = new EventProvider({ url: 'sample.com' });
|
||||
const events = provider._createEvents(sample.events);
|
||||
const eventObj = {
|
||||
eventType: 'print',
|
||||
displayTime: '00:45',
|
||||
ms: 4523,
|
||||
bytes: 6516,
|
||||
offset: 137723,
|
||||
data: null,
|
||||
w: 115,
|
||||
h: 23,
|
||||
time: new Date('2016-05-09T14:57:51.238Z'),
|
||||
msNormalized: 1744,
|
||||
};
|
||||
|
||||
expect(events).toHaveLength(32);
|
||||
expect(events[30]).toEqual(eventObj);
|
||||
});
|
||||
|
||||
it('should fetch session content', async () => {
|
||||
const provider = new EventProvider({ url: 'sample.com' });
|
||||
|
||||
jest
|
||||
.spyOn(provider, '_fetchEvents')
|
||||
.mockImplementation(() =>
|
||||
Promise.resolve(provider._createEvents(sample.events))
|
||||
);
|
||||
|
||||
jest
|
||||
.spyOn(api, 'fetch')
|
||||
.mockImplementation(() => Promise.resolve({ text: () => sample.data }));
|
||||
|
||||
await provider.init();
|
||||
|
||||
expect(api.fetch).toHaveBeenCalledWith(
|
||||
`sample.com/stream?offset=0&bytes=${MAX_SIZE}`,
|
||||
{
|
||||
Accept: 'text/plain',
|
||||
'Content-Type': 'text/plain; charset=utf-8',
|
||||
}
|
||||
);
|
||||
|
||||
const buf = new Buffer(sample.data);
|
||||
const lastEvent = provider.events[provider.events.length - 2];
|
||||
const expectedChunk = buf
|
||||
.slice(lastEvent.offset, lastEvent.offset + lastEvent.bytes)
|
||||
.toString('utf8');
|
||||
expect(lastEvent.data).toEqual(expectedChunk);
|
||||
});
|
||||
});
|
||||
import TtyPlayer, { StatusEnum } from './ttyPlayer';
|
||||
|
||||
describe('lib/ttyPlayer', () => {
|
||||
let server;
|
||||
const url = 'ws://localhost:3088';
|
||||
beforeEach(() => {
|
||||
server = new WS(url);
|
||||
});
|
||||
afterEach(() => {
|
||||
WS.clean();
|
||||
|
||||
jest.clearAllMocks();
|
||||
});
|
||||
|
||||
it('should create an instance', () => {
|
||||
const ttyPlayer = new TtyPlayer({ url: 'testSid' });
|
||||
expect(ttyPlayer.isReady()).toBe(false);
|
||||
expect(ttyPlayer.isPlaying()).toBe(false);
|
||||
expect(ttyPlayer.isError()).toBe(false);
|
||||
expect(ttyPlayer.isLoading()).toBe(true);
|
||||
expect(ttyPlayer.duration).toBe(0);
|
||||
expect(ttyPlayer.current).toBe(0);
|
||||
});
|
||||
it('connects to a websocket', async () => {
|
||||
const setPlayerStatus = jest.fn();
|
||||
const setStatusText = () => {};
|
||||
const setTime = () => {};
|
||||
|
||||
it('should connect using event provider', async () => {
|
||||
const ttyPlayer = new TtyPlayer(new EventProvider({ url: 'testSid' }));
|
||||
|
||||
jest.spyOn(api, 'get').mockImplementation(() => Promise.resolve(sample));
|
||||
jest
|
||||
.spyOn(ttyPlayer._eventProvider, '_fetchContent')
|
||||
.mockImplementation(() => Promise.resolve(sample.data));
|
||||
|
||||
await ttyPlayer.connect();
|
||||
|
||||
expect(ttyPlayer.isReady()).toBe(true);
|
||||
expect(ttyPlayer.getEventCount()).toBe(32);
|
||||
});
|
||||
|
||||
it('should indicate its loading status', async () => {
|
||||
const ttyPlayer = new TtyPlayer(new EventProvider({ url: 'testSid' }));
|
||||
jest
|
||||
.spyOn(api, 'get')
|
||||
.mockImplementation(() => Promise.resolve({ events: [] }));
|
||||
const ttyPlayer = new TtyPlayer({
|
||||
url,
|
||||
setPlayerStatus,
|
||||
setStatusText,
|
||||
setTime,
|
||||
});
|
||||
const emit = jest.spyOn(ttyPlayer, 'emit');
|
||||
|
||||
ttyPlayer.connect();
|
||||
expect(ttyPlayer.isLoading()).toBe(true);
|
||||
await server.connected;
|
||||
|
||||
expect(setPlayerStatus.mock.calls).toHaveLength(1);
|
||||
expect(setPlayerStatus.mock.calls[0][0]).toBe(StatusEnum.LOADING);
|
||||
|
||||
expect(emit.mock.calls).toHaveLength(1);
|
||||
expect(emit.mock.calls[0][0]).toBe('open');
|
||||
|
||||
server.close();
|
||||
await server.closed;
|
||||
|
||||
expect(emit.mock.calls).toHaveLength(2);
|
||||
expect(emit.mock.calls[1][0]).toBe(TermEvent.CONN_CLOSE);
|
||||
});
|
||||
|
||||
it('should indicate its error status', async () => {
|
||||
jest.spyOn(console, 'error').mockImplementation(() => {});
|
||||
jest.spyOn(api, 'get').mockImplementation(() => Promise.reject('!!'));
|
||||
it('emits resize events', async () => {
|
||||
const setPlayerStatus = () => {};
|
||||
const setStatusText = () => {};
|
||||
const setTime = () => {};
|
||||
|
||||
const ttyPlayer = new TtyPlayer(new EventProvider({ url: 'testSid' }));
|
||||
const ttyPlayer = new TtyPlayer({
|
||||
url,
|
||||
setPlayerStatus,
|
||||
setStatusText,
|
||||
setTime,
|
||||
});
|
||||
const emit = jest.spyOn(ttyPlayer, 'emit');
|
||||
|
||||
await ttyPlayer.connect();
|
||||
expect(ttyPlayer.isError()).toBe(true);
|
||||
ttyPlayer.connect();
|
||||
await server.connected;
|
||||
|
||||
const resizeMessage = new Uint8Array([
|
||||
5, // message type = Resize
|
||||
0,
|
||||
4, // size
|
||||
0,
|
||||
80, // width
|
||||
0,
|
||||
60, // height
|
||||
]);
|
||||
|
||||
server.send(resizeMessage.buffer);
|
||||
|
||||
expect(emit.mock.lastCall).toBeDefined();
|
||||
expect(emit.mock.lastCall[0]).toBe(TermEvent.RESIZE);
|
||||
expect(emit.mock.lastCall[1]).toStrictEqual({ w: 80, h: 60 });
|
||||
});
|
||||
|
||||
describe('move()', () => {
|
||||
var tty = null;
|
||||
it('plays PTY data', async () => {
|
||||
const setPlayerStatus = jest.fn();
|
||||
const setStatusText = jest.fn();
|
||||
const setTime = jest.fn();
|
||||
|
||||
beforeEach(() => {
|
||||
tty = new TtyPlayer(new EventProvider({ url: 'testSid' }));
|
||||
jest.spyOn(api, 'get').mockImplementation(() => Promise.resolve(sample));
|
||||
jest
|
||||
.spyOn(tty._eventProvider, '_fetchContent')
|
||||
.mockImplementation(() => Promise.resolve(sample.data));
|
||||
const ttyPlayer = new TtyPlayer({
|
||||
url,
|
||||
setPlayerStatus,
|
||||
setStatusText,
|
||||
setTime,
|
||||
});
|
||||
const emit = jest.spyOn(ttyPlayer, 'emit');
|
||||
|
||||
afterEach(function () {
|
||||
jest.clearAllMocks();
|
||||
});
|
||||
ttyPlayer.connect();
|
||||
await server.connected;
|
||||
|
||||
it('should move by 1 position when called w/o params', async () => {
|
||||
await tty.connect();
|
||||
const data = new TextEncoder('utf-8').encode('~/test $');
|
||||
const len = data.length + 8;
|
||||
const ptyMessage = new Uint8Array([
|
||||
1 /* message type = PTY */,
|
||||
len >> 8,
|
||||
len & 0xff /* length */,
|
||||
0,
|
||||
0,
|
||||
0,
|
||||
0,
|
||||
0,
|
||||
0,
|
||||
0,
|
||||
123 /* timestamp (123ms) */,
|
||||
...data,
|
||||
]);
|
||||
|
||||
let renderedData = '';
|
||||
tty.on(TermEvent.DATA, data => {
|
||||
renderedData = data;
|
||||
});
|
||||
server.send(ptyMessage.buffer);
|
||||
|
||||
tty.move();
|
||||
expect(renderedData).toHaveLength(42);
|
||||
});
|
||||
expect(emit.mock.lastCall).toBeDefined();
|
||||
expect(emit.mock.lastCall[0]).toBe(TermEvent.DATA);
|
||||
|
||||
it('should move from 1 to 478 position (forward)', async () => {
|
||||
await tty.connect();
|
||||
|
||||
const renderedDataLength = [];
|
||||
const resizeEvents = [];
|
||||
|
||||
tty.on(TermEvent.RESIZE, event => {
|
||||
resizeEvents.push(event);
|
||||
});
|
||||
|
||||
tty.on(TermEvent.DATA, data => {
|
||||
renderedDataLength.push(data.length);
|
||||
tty.resumeFlow();
|
||||
});
|
||||
|
||||
tty.move(478);
|
||||
|
||||
const expected = [
|
||||
{
|
||||
resize: {
|
||||
h: 20,
|
||||
w: 147,
|
||||
},
|
||||
length: 12899,
|
||||
},
|
||||
{
|
||||
resize: {
|
||||
h: 29,
|
||||
w: 146,
|
||||
},
|
||||
length: 9415,
|
||||
},
|
||||
{
|
||||
resize: {
|
||||
h: 31,
|
||||
w: 146,
|
||||
},
|
||||
length: 10113,
|
||||
},
|
||||
{
|
||||
resize: {
|
||||
h: 25,
|
||||
w: 146,
|
||||
},
|
||||
length: 8018,
|
||||
},
|
||||
];
|
||||
|
||||
for (let i = 0; i < expected.length; i++) {
|
||||
expect(resizeEvents[i]).toEqual(expected[i].resize);
|
||||
expect(renderedDataLength[i]).toBe(expected[i].length);
|
||||
}
|
||||
});
|
||||
|
||||
it('should move from 478 to 1 position (back)', async () => {
|
||||
await tty.connect();
|
||||
|
||||
let renderedData = '';
|
||||
tty.current = 478;
|
||||
tty.on(TermEvent.DATA, data => {
|
||||
renderedData = data;
|
||||
});
|
||||
|
||||
tty.move(2);
|
||||
expect(renderedData).toHaveLength(42);
|
||||
});
|
||||
|
||||
it('should stop playing if new position is greater than session length', async () => {
|
||||
await tty.connect();
|
||||
tty.play();
|
||||
|
||||
const someBigNumber = 20000;
|
||||
tty.move(someBigNumber);
|
||||
|
||||
expect(tty.isPlaying()).toBe(false);
|
||||
});
|
||||
expect(emit.mock.lastCall[1]).toStrictEqual(Uint8Array.from(data).buffer);
|
||||
expect(setTime.mock.lastCall).toBeDefined();
|
||||
expect(setTime.mock.lastCall[0]).toBe(123);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -1,230 +0,0 @@
|
||||
/*
|
||||
* Teleport
|
||||
* Copyright (C) 2023 Gravitational, Inc.
|
||||
*
|
||||
* This program is free software: you can redistribute it and/or modify
|
||||
* it under the terms of the GNU Affero General Public License as published by
|
||||
* the Free Software Foundation, either version 3 of the License, or
|
||||
* (at your option) any later version.
|
||||
*
|
||||
* This program is distributed in the hope that it will be useful,
|
||||
* but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
* GNU Affero General Public License for more details.
|
||||
*
|
||||
* You should have received a copy of the GNU Affero General Public License
|
||||
* along with this program. If not, see <http://www.gnu.org/licenses/>.
|
||||
*/
|
||||
|
||||
import BufferModule from 'buffer/';
|
||||
|
||||
import api from 'teleport/services/api';
|
||||
|
||||
import { EventType } from './enums';
|
||||
|
||||
const URL_PREFIX_EVENTS = '/events';
|
||||
const Buffer = BufferModule.Buffer;
|
||||
|
||||
export const MAX_SIZE = 5242880; // 5mg
|
||||
|
||||
export default class EventProvider {
|
||||
constructor({ url }) {
|
||||
this.url = url;
|
||||
this.events = [];
|
||||
}
|
||||
|
||||
getDuration() {
|
||||
const eventCount = this.events.length;
|
||||
if (eventCount === 0) {
|
||||
return 0;
|
||||
}
|
||||
|
||||
return this.events[eventCount - 1].msNormalized;
|
||||
}
|
||||
|
||||
init() {
|
||||
return this._fetchEvents().then(events => {
|
||||
this.events = events;
|
||||
const printEvents = this.events.filter(onlyPrintEvents);
|
||||
if (printEvents.length === 0) {
|
||||
return;
|
||||
}
|
||||
|
||||
return this._fetchContent(printEvents).then(buffer => {
|
||||
this._populatePrintEvents(buffer, printEvents);
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
_fetchEvents() {
|
||||
const url = this.url + URL_PREFIX_EVENTS;
|
||||
return api.get(url).then(json => {
|
||||
if (!json.events) {
|
||||
return [];
|
||||
}
|
||||
|
||||
return this._createEvents(json.events);
|
||||
});
|
||||
}
|
||||
|
||||
_fetchContent(events) {
|
||||
// calculate the size of the session in bytes to know how many
|
||||
// chunks to load due to maximum chunk size limitation.
|
||||
let offset = events[0].offset;
|
||||
const end = events.length - 1;
|
||||
const totalSize = events[end].offset - offset + events[end].bytes;
|
||||
const chunkCount = Math.ceil(totalSize / MAX_SIZE);
|
||||
|
||||
// create a fetch request for each chunk
|
||||
const promises = [];
|
||||
for (let i = 0; i < chunkCount; i++) {
|
||||
const url = `${this.url}/stream?offset=${offset}&bytes=${MAX_SIZE}`;
|
||||
promises.push(
|
||||
api
|
||||
.fetch(url, {
|
||||
Accept: 'text/plain',
|
||||
'Content-Type': 'text/plain; charset=utf-8',
|
||||
})
|
||||
.then(response => response.text())
|
||||
);
|
||||
offset = offset + MAX_SIZE;
|
||||
}
|
||||
|
||||
// fetch all chunks and then merge
|
||||
return Promise.all(promises).then(responses => {
|
||||
const allBytes = responses.reduce((byteStr, r) => byteStr + r, '');
|
||||
return new Buffer(allBytes);
|
||||
});
|
||||
}
|
||||
|
||||
// assign a slice of tty stream to corresponding print event
|
||||
_populatePrintEvents(buffer, events) {
|
||||
let byteStrOffset = events[0].bytes;
|
||||
events[0].data = buffer.slice(0, byteStrOffset).toString('utf8');
|
||||
for (var i = 1; i < events.length; i++) {
|
||||
let { bytes } = events[i];
|
||||
events[i].data = buffer
|
||||
.slice(byteStrOffset, byteStrOffset + bytes)
|
||||
.toString('utf8');
|
||||
byteStrOffset += bytes;
|
||||
}
|
||||
}
|
||||
|
||||
_createEvents(json) {
|
||||
let w, h;
|
||||
let events = [];
|
||||
|
||||
// filter print events and ensure that each has the right screen size and valid values
|
||||
for (let i = 0; i < json.length; i++) {
|
||||
const { ms, event, offset, time, bytes } = json[i];
|
||||
|
||||
// grab new screen size for the next events
|
||||
if (event === EventType.RESIZE || event === EventType.START) {
|
||||
[w, h] = json[i].size.split(':');
|
||||
}
|
||||
|
||||
// session has ended, stop here
|
||||
if (event === EventType.END) {
|
||||
const start = new Date(events[0].time);
|
||||
const end = new Date(time);
|
||||
const duration = end.getTime() - start.getTime();
|
||||
events.push({
|
||||
eventType: event,
|
||||
ms: duration,
|
||||
time: new Date(time),
|
||||
});
|
||||
|
||||
break;
|
||||
}
|
||||
|
||||
// process only PRINT events
|
||||
if (event !== EventType.PRINT) {
|
||||
continue;
|
||||
}
|
||||
|
||||
events.push({
|
||||
eventType: EventType.PRINT,
|
||||
ms,
|
||||
bytes,
|
||||
offset,
|
||||
data: null,
|
||||
w: Number(w),
|
||||
h: Number(h),
|
||||
time: new Date(time),
|
||||
});
|
||||
}
|
||||
|
||||
return this._normalizeEventsByTime(events);
|
||||
}
|
||||
|
||||
_normalizeEventsByTime(events) {
|
||||
if (!events || events.length === 0) {
|
||||
return [];
|
||||
}
|
||||
|
||||
events.forEach(e => {
|
||||
e.displayTime = formatDisplayTime(e.ms);
|
||||
e.ms = e.ms > 0 ? Math.floor(e.ms / 10) : 0;
|
||||
e.msNormalized = e.ms;
|
||||
});
|
||||
|
||||
let cur = events[0];
|
||||
let tmp = [];
|
||||
for (let i = 1; i < events.length; i++) {
|
||||
const sameSize = cur.w === events[i].w && cur.h === events[i].h;
|
||||
const delay = events[i].ms - cur.ms;
|
||||
|
||||
// merge events with tiny delay
|
||||
if (delay < 2 && sameSize) {
|
||||
cur.bytes += events[i].bytes;
|
||||
continue;
|
||||
}
|
||||
|
||||
// avoid long delays between chunks
|
||||
events[i].msNormalized = cur.msNormalized + shortenTime(delay);
|
||||
|
||||
tmp.push(cur);
|
||||
cur = events[i];
|
||||
}
|
||||
|
||||
if (tmp.indexOf(cur) === -1) {
|
||||
tmp.push(cur);
|
||||
}
|
||||
|
||||
return tmp;
|
||||
}
|
||||
}
|
||||
|
||||
function shortenTime(value) {
|
||||
if (value >= 25 && value < 50) {
|
||||
return 25;
|
||||
} else if (value >= 50 && value < 100) {
|
||||
return 50;
|
||||
} else if (value >= 100) {
|
||||
return 100;
|
||||
} else {
|
||||
return value;
|
||||
}
|
||||
}
|
||||
|
||||
function formatDisplayTime(ms) {
|
||||
if (ms <= 0) {
|
||||
return '00:00';
|
||||
}
|
||||
|
||||
let totalSec = Math.floor(ms / 1000);
|
||||
let totalDays = (totalSec % 31536000) % 86400;
|
||||
let h = Math.floor(totalDays / 3600);
|
||||
let m = Math.floor((totalDays % 3600) / 60);
|
||||
let s = (totalDays % 3600) % 60;
|
||||
|
||||
m = m > 9 ? m : '0' + m;
|
||||
s = s > 9 ? s : '0' + s;
|
||||
h = h > 0 ? h + ':' : '';
|
||||
|
||||
return `${h}${m}:${s}`;
|
||||
}
|
||||
|
||||
export function onlyPrintEvents(e) {
|
||||
return e.eventType === EventType.PRINT;
|
||||
}
|
||||
@@ -10862,7 +10862,7 @@ jest-config@^29.7.0:
|
||||
slash "^3.0.0"
|
||||
strip-json-comments "^3.1.1"
|
||||
|
||||
jest-diff@^29.7.0:
|
||||
jest-diff@^29.2.0, jest-diff@^29.7.0:
|
||||
version "29.7.0"
|
||||
resolved "https://registry.yarnpkg.com/jest-diff/-/jest-diff-29.7.0.tgz#017934a66ebb7ecf6f205e84699be10afd70458a"
|
||||
integrity sha512-LMIgiIrhigmPrs03JHpxUh2yISK3vLFPkAodPeo0+BuF7wA2FoQbkEg1u8gBYBThncu7e1oEDUfIXVuTqLRUjw==
|
||||
@@ -11141,6 +11141,14 @@ jest-watcher@^29.7.0:
|
||||
jest-util "^29.7.0"
|
||||
string-length "^4.0.1"
|
||||
|
||||
jest-websocket-mock@^2.5.0:
|
||||
version "2.5.0"
|
||||
resolved "https://registry.yarnpkg.com/jest-websocket-mock/-/jest-websocket-mock-2.5.0.tgz#9e0b07e270bed0224a6d3269fc62625eaa4d465c"
|
||||
integrity sha512-a+UJGfowNIWvtIKIQBHoEWIUqRxxQHFx4CXT+R5KxxKBtEQ5rS3pPOV/5299sHzqbmeCzxxY5qE4+yfXePePig==
|
||||
dependencies:
|
||||
jest-diff "^29.2.0"
|
||||
mock-socket "^9.3.0"
|
||||
|
||||
jest-worker@^26.5.0:
|
||||
version "26.6.2"
|
||||
resolved "https://registry.yarnpkg.com/jest-worker/-/jest-worker-26.6.2.tgz#7f72cbc4d643c365e27b9fd775f9d0eaa9c7a8ed"
|
||||
@@ -12499,6 +12507,11 @@ mkdirp@^1.0.3, mkdirp@^1.0.4:
|
||||
resolved "https://registry.yarnpkg.com/mkdirp/-/mkdirp-1.0.4.tgz#3eb5ed62622756d79a5f0e2a221dfebad75c2f7e"
|
||||
integrity sha512-vVqVZQyf3WLx2Shd0qJ9xuvqgAyKPLAiqITEtqW0oIUjzo3PePDd6fW9iFz30ef7Ysp/oiWqbhszeGWW2T6Gzw==
|
||||
|
||||
mock-socket@^9.3.0:
|
||||
version "9.3.1"
|
||||
resolved "https://registry.yarnpkg.com/mock-socket/-/mock-socket-9.3.1.tgz#24fb00c2f573c84812aa4a24181bb025de80cc8e"
|
||||
integrity sha512-qxBgB7Qa2sEQgHFjj0dSigq7fX4k6Saisd5Nelwp2q8mlbAFh5dHV9JTTlF8viYJLSSWgMCZFUom8PJcMNBoJw==
|
||||
|
||||
module-details-from-path@^1.0.3:
|
||||
version "1.0.3"
|
||||
resolved "https://registry.yarnpkg.com/module-details-from-path/-/module-details-from-path-1.0.3.tgz#114c949673e2a8a35e9d35788527aa37b679da2b"
|
||||
@@ -16848,9 +16861,9 @@ webpack-virtual-modules@^0.4.1:
|
||||
integrity sha512-5NUqC2JquIL2pBAAo/VfBP6KuGkHIZQXW/lNKupLPfhViwh8wNsu0BObtl09yuKZszeEUfbXz8xhrHvSG16Nqw==
|
||||
|
||||
webpack@4, "webpack@>=4.43.0 <6.0.0", webpack@^5, webpack@^5.76.2, webpack@^5.88.2, webpack@^5.9.0:
|
||||
version "5.88.2"
|
||||
resolved "https://registry.yarnpkg.com/webpack/-/webpack-5.88.2.tgz#f62b4b842f1c6ff580f3fcb2ed4f0b579f4c210e"
|
||||
integrity sha512-JmcgNZ1iKj+aiR0OvTYtWQqJwq37Pf683dY9bVORwVbUrDhLhdn/PlO2sHsFHPkj7sHNQF3JwaAkp49V+Sq1tQ==
|
||||
version "5.89.0"
|
||||
resolved "https://registry.yarnpkg.com/webpack/-/webpack-5.89.0.tgz#56b8bf9a34356e93a6625770006490bf3a7f32dc"
|
||||
integrity sha512-qyfIC10pOr70V+jkmud8tMfajraGCZMBWJtrmuBymQKCrLTRejBI8STDp1MCyZu/QTdZSeacCQYpYNQVOzX5kw==
|
||||
dependencies:
|
||||
"@types/eslint-scope" "^3.7.3"
|
||||
"@types/estree" "^1.0.0"
|
||||
|
||||
Reference in New Issue
Block a user