stroblme
d342d923c6
A port with `stream` set publishes repeatedly during one execution, and until now nothing on the canvas said so. `deriveEdges` carries the flag from the producing `MessageSpec` onto the edge, `LiveEdge` turns it into an `edge-stream` class, and the class draws a dashed stroke whose offset marches toward the target. The pulse still lands on top: both animations share one declaration when a value arrives on a streaming edge.
92 lines
2.7 KiB
TypeScript
92 lines
2.7 KiB
TypeScript
import {
|
|
BaseEdge,
|
|
EdgeLabelRenderer,
|
|
type EdgeProps,
|
|
getBezierPath,
|
|
useStore,
|
|
} from "@xyflow/react"
|
|
import { memo, useEffect, useRef, useState } from "react"
|
|
|
|
import { duration } from "@/lib/motion"
|
|
import { cn } from "@/lib/utils"
|
|
import type { FlowEdgeData } from "./deriveEdges"
|
|
import { useLiveValue } from "./liveStore"
|
|
|
|
/** Below this zoom the value chips would be unreadable, so they step aside. */
|
|
const CHIP_MIN_ZOOM = 0.5
|
|
|
|
function formatValue(value: unknown): string {
|
|
if (typeof value === "number") {
|
|
return Number.isInteger(value) ? String(value) : value.toFixed(2)
|
|
}
|
|
if (typeof value === "string") return value
|
|
return JSON.stringify(value) ?? ""
|
|
}
|
|
|
|
function LiveEdgeComponent({
|
|
id,
|
|
sourceX,
|
|
sourceY,
|
|
targetX,
|
|
targetY,
|
|
sourcePosition,
|
|
targetPosition,
|
|
data,
|
|
selected,
|
|
}: EdgeProps) {
|
|
const { message, producerId, stream } = (data ?? {}) as FlowEdgeData
|
|
const live = useLiveValue(message)
|
|
const zoom = useStore((state) => state.transform[2])
|
|
const [pulsing, setPulsing] = useState(false)
|
|
const lastTs = useRef<number | null>(null)
|
|
|
|
const [path, labelX, labelY] = getBezierPath({
|
|
sourceX,
|
|
sourceY,
|
|
sourcePosition,
|
|
targetX,
|
|
targetY,
|
|
targetPosition,
|
|
})
|
|
|
|
// Restart the stroke animation whenever a newer message lands — but only
|
|
// for the producer that actually published it. A message can have several
|
|
// producers, and can also be set from a dashboard or another flow, so
|
|
// pulsing on the value alone claims things happened that did not.
|
|
useEffect(() => {
|
|
if (!live?.ts || live.ts === lastTs.current) return
|
|
lastTs.current = live.ts
|
|
const from = live.source
|
|
// No source at all is an older engine; pulse rather than go silent.
|
|
if (from && from.id !== producerId) return
|
|
setPulsing(true)
|
|
const timer = setTimeout(() => setPulsing(false), duration.pulse * 1000)
|
|
return () => clearTimeout(timer)
|
|
}, [live?.ts, live?.source, producerId])
|
|
|
|
return (
|
|
<>
|
|
<BaseEdge
|
|
id={id}
|
|
path={path}
|
|
className={cn(pulsing && "edge-live", stream && "edge-stream")}
|
|
style={selected ? { stroke: "var(--primary)" } : undefined}
|
|
/>
|
|
{live !== undefined && zoom >= CHIP_MIN_ZOOM ? (
|
|
<EdgeLabelRenderer>
|
|
<div
|
|
style={{
|
|
transform: `translate(-50%, -50%) translate(${labelX}px, ${labelY}px)`,
|
|
}}
|
|
className="pointer-events-none absolute max-w-[140px] truncate rounded-full border border-border bg-card px-2 py-0.5 font-mono text-xs text-foreground"
|
|
>
|
|
{formatValue(live.value)}
|
|
</div>
|
|
</EdgeLabelRenderer>
|
|
) : null}
|
|
</>
|
|
)
|
|
}
|
|
|
|
export const LiveEdge = memo(LiveEdgeComponent)
|