Time-based Operators and Temporal Logic

import { fromEvent, interval } from 'rxjs';
import { debounceTime, map, distinctUntilChanged } from 'rxjs/operators';

// Basic debounceTime - wait for quiet period
const input$ = fromEvent(searchInput, 'input');

input$.pipe(
  map(e => e.target.value),
  debounceTime(300), // Wait 300ms after typing stops
  distinctUntilChanged()
).subscribe(searchTerm => {
  console.log('Search for:', searchTerm);
  performSearch(searchTerm);
});

// Rapid emissions - only last emitted
interval(100).pipe(
  take(10),
  debounceTime(500)
).subscribe(val => console.log('Debounced:', val));
// Only emits value 9 (after 500ms silence)

// Practical: Type-ahead search
fromEvent(searchBox, 'input').pipe(
  map(e => e.target.value),
  debounceTime(400), // Wait 400ms after user stops typing
  distinctUntilChanged(),
  filter(term => term.length >= 2),
  switchMap(term => ajax.getJSON(`/api/search?q=${term}`))
).subscribe(results => {
  displaySearchResults(results);
  // Reduces API calls dramatically
});

// Practical: Auto-save form
const formChanges$ = merge(
  fromEvent(input1, 'input'),
  fromEvent(input2, 'input'),
  fromEvent(textarea, 'input')
);

formChanges$.pipe(
  debounceTime(2000), // Save 2s after user stops editing
  map(() => getFormData()),
  distinctUntilChanged((prev, curr) => 
    JSON.stringify(prev) === JSON.stringify(curr)
  ),
  switchMap(data => ajax.post('/api/save', data))
).subscribe({
  next: () => showSavedIndicator(),
  error: err => showSaveError(err)
});

// Practical: Window resize handling
fromEvent(window, 'resize').pipe(
  debounceTime(250), // Wait 250ms after resize stops
  map(() => ({
    width: window.innerWidth,
    height: window.innerHeight
  }))
).subscribe(dimensions => {
  recalculateLayout(dimensions);
  repositionElements(dimensions);
  // Prevents excessive layout calculations
});

// Practical: Scroll end detection
fromEvent(window, 'scroll').pipe(
  debounceTime(150), // User stopped scrolling
  map(() => window.scrollY)
).subscribe(scrollPosition => {
  saveScrollPosition(scrollPosition);
  checkIfAtBottom(scrollPosition);
});

// Practical: Live validation with delay
fromEvent(emailInput, 'input').pipe(
  map(e => e.target.value),
  debounceTime(600),
  distinctUntilChanged(),
  switchMap(email => validateEmail(email))
).subscribe(validation => {
  if (validation.valid) {
    showValidIcon();
  } else {
    showInvalidIcon(validation.errors);
  }
});

// Practical: Infinite scroll trigger
fromEvent(window, 'scroll').pipe(
  debounceTime(100),
  filter(() => isNearBottom()),
  switchMap(() => loadMoreItems())
).subscribe(items => appendItems(items));

// Practical: Text editor auto-complete
fromEvent(editor, 'input').pipe(
  map(e => ({
    text: e.target.value,
    cursorPosition: e.target.selectionStart
  })),
  debounceTime(300),
  filter(state => shouldShowAutoComplete(state)),
  switchMap(state => getSuggestions(state.text))
).subscribe(suggestions => displayAutoComplete(suggestions));

// Practical: Real-time collaboration presence
const userActivity$ = merge(
  fromEvent(document, 'mousemove'),
  fromEvent(document, 'keypress'),
  fromEvent(document, 'click')
);

userActivity$.pipe(
  debounceTime(5000), // 5s of inactivity
  mapTo('idle')
).subscribe(status => {
  updatePresenceStatus(status);
  sendStatusToServer(status);
});

// Practical: Filter panel updates
const filterChanges$ = merge(
  fromEvent(priceRange, 'input'),
  fromEvent(categorySelect, 'change'),
  fromEvent(ratingFilter, 'change')
);

filterChanges$.pipe(
  debounceTime(500),
  map(() => getFilterValues()),
  distinctUntilChanged((a, b) => deepEqual(a, b)),
  switchMap(filters => ajax.post('/api/filter', filters))
).subscribe(results => updateProductList(results));

// Practical: Chat typing indicator
fromEvent(chatInput, 'input').pipe(
  debounceTime(1000),
  mapTo(false) // Stopped typing
).subscribe(isTyping => {
  sendTypingStatus(isTyping);
});

// With immediate emission on first input
fromEvent(chatInput, 'input').pipe(
  tap(() => sendTypingStatus(true)), // Immediate
  debounceTime(1000),
  tap(() => sendTypingStatus(false)) // After stop
).subscribe();

// Comparison: debounce vs throttle
// debounceTime: Waits for silence, emits last
fromEvent(button, 'click').pipe(
  debounceTime(1000)
).subscribe(() => console.log('Debounced click'));

// throttleTime: Emits first, ignores until period ends
fromEvent(button, 'click').pipe(
  throttleTime(1000)
).subscribe(() => console.log('Throttled click'));

import { fromEvent, interval } from 'rxjs';
import { throttleTime, map, scan } from 'rxjs/operators';

// Basic throttleTime - emit first, ignore rest
fromEvent(button, 'click').pipe(
  throttleTime(1000)
).subscribe(() => console.log('Throttled click'));
// First click processed, others ignored for 1s

// Rapid emissions throttled
interval(100).pipe(
  take(20),
  throttleTime(1000)
).subscribe(val => console.log('Throttled:', val));
// Emits: 0, 10 (approx)

// Trailing edge configuration
fromEvent(button, 'click').pipe(
  throttleTime(1000, asyncScheduler, { 
    leading: false, 
    trailing: true 
  })
).subscribe(() => console.log('Trailing throttle'));

// Practical: Scroll event throttling
fromEvent(window, 'scroll').pipe(
  throttleTime(200), // Max 5 times per second
  map(() => window.scrollY)
).subscribe(scrollY => {
  updateScrollProgress(scrollY);
  toggleBackToTop(scrollY);
  // Prevents excessive calculations
});

// Practical: Button click protection
fromEvent(submitButton, 'click').pipe(
  throttleTime(2000) // Prevent double-click
).subscribe(() => {
  console.log('Processing submission...');
  submitForm();
});

// Practical: Mouse move tracking (performance)
fromEvent(canvas, 'mousemove').pipe(
  throttleTime(50), // ~20 updates per second
  map(e => ({ x: e.clientX, y: e.clientY }))
).subscribe(position => {
  updateCursor(position);
  drawAtPosition(position);
});

// Practical: Window resize (immediate response)
fromEvent(window, 'resize').pipe(
  throttleTime(100) // Leading edge - immediate first response
).subscribe(() => {
  updateResponsiveLayout();
  // First resize handled immediately, then throttled
});

// Practical: API rate limiting
const apiCalls$ = new Subject<Request>();

apiCalls$.pipe(
  throttleTime(1000), // Max 1 call per second
  switchMap(request => ajax(request))
).subscribe(response => handleResponse(response));

// Queue multiple calls
apiCalls$.next(request1);
apiCalls$.next(request2); // Ignored
apiCalls$.next(request3); // Ignored
// Only request1 processed

// Practical: Save button rate limiting
fromEvent(saveBtn, 'click').pipe(
  throttleTime(3000),
  switchMap(() => saveData())
).subscribe({
  next: () => showSaveSuccess(),
  error: err => showSaveError(err)
});

// Practical: Infinite scroll throttling
fromEvent(window, 'scroll').pipe(
  throttleTime(300),
  filter(() => isNearBottom()),
  exhaustMap(() => loadNextPage())
).subscribe(items => appendItems(items));

// Practical: Keyboard event throttling
fromEvent(document, 'keydown').pipe(
  throttleTime(100),
  map(e => e.key)
).subscribe(key => {
  handleKeyPress(key);
  // Prevents key repeat spam
});

// Practical: Analytics event throttling
const trackEvent = (category: string, action: string) => {
  return of({ category, action }).pipe(
    throttleTime(5000) // Max once per 5 seconds
  ).subscribe(event => {
    analytics.track(event.category, event.action);
  });
};

// Practical: Drag and drop throttling
fromEvent(element, 'drag').pipe(
  throttleTime(16), // ~60fps
  map(e => ({ x: e.clientX, y: e.clientY }))
).subscribe(position => {
  updateDragPreview(position);
});

// Both leading and trailing
fromEvent(input, 'input').pipe(
  throttleTime(1000, asyncScheduler, {
    leading: true,  // Immediate first
    trailing: true  // Last after period
  }),
  map(e => e.target.value)
).subscribe(value => processValue(value));

// Practical: Network request throttling
const requests$ = new Subject<string>();

requests$.pipe(
  throttleTime(200),
  switchMap(url => ajax.getJSON(url))
).subscribe(data => updateUI(data));

// Practical: Game input throttling
fromEvent(document, 'keypress').pipe(
  filter(e => e.key === ' '), // Spacebar
  throttleTime(500) // Limit firing rate
).subscribe(() => {
  fireWeapon();
});

// Practical: Audio visualization throttling
audioData$.pipe(
  throttleTime(33) // ~30fps for visualization
).subscribe(data => {
  updateVisualization(data);
});

import { fromEvent, interval } from 'rxjs';
import { auditTime, map } from 'rxjs/operators';

// Basic auditTime - emit last value after period
fromEvent(button, 'click').pipe(
  auditTime(1000)
).subscribe(() => console.log('Audit: last click after 1s'));

// Rapid emissions - emits last
interval(100).pipe(
  take(15),
  auditTime(1000)
).subscribe(val => console.log('Audited:', val));
// Emits values at trailing edge

// Practical: Capture final scroll position
fromEvent(window, 'scroll').pipe(
  auditTime(200),
  map(() => window.scrollY)
).subscribe(finalScrollY => {
  saveScrollPosition(finalScrollY);
  updateSectionHighlight(finalScrollY);
  // Uses final position after scroll settles
});

// Practical: Final resize dimensions
fromEvent(window, 'resize').pipe(
  auditTime(300),
  map(() => ({
    width: window.innerWidth,
    height: window.innerHeight
  }))
).subscribe(finalDimensions => {
  recalculateLayout(finalDimensions);
  // Uses final window size
});

// Practical: Input field final value
fromEvent(input, 'input').pipe(
  map(e => e.target.value),
  auditTime(500)
).subscribe(finalValue => {
  console.log('Final value:', finalValue);
  performValidation(finalValue);
});

// Practical: Mouse position after movement stops
fromEvent(canvas, 'mousemove').pipe(
  auditTime(200),
  map(e => ({ x: e.clientX, y: e.clientY }))
).subscribe(finalPosition => {
  showContextMenuAt(finalPosition);
  // Shows menu at final mouse position
});

// Practical: Slider final value
fromEvent(slider, 'input').pipe(
  map(e => e.target.value),
  auditTime(300)
).subscribe(finalValue => {
  applyFilterWithValue(finalValue);
  // Applies filter after user stops adjusting
});

// Practical: Final drag position
fromEvent(draggable, 'drag').pipe(
  auditTime(100),
  map(e => ({ x: e.clientX, y: e.clientY }))
).subscribe(finalPosition => {
  snapToGrid(finalPosition);
  savePosition(finalPosition);
});

// Compare timing operators
const clicks$ = fromEvent(button, 'click');

// throttleTime: First click immediately
clicks$.pipe(throttleTime(1000))
  .subscribe(() => console.log('Throttle: first'));

// auditTime: Last click after period
clicks$.pipe(auditTime(1000))
  .subscribe(() => console.log('Audit: last'));

// debounceTime: After clicking stops
clicks$.pipe(debounceTime(1000))
  .subscribe(() => console.log('Debounce: after stop'));

// Practical: Color picker final selection
fromEvent(colorPicker, 'input').pipe(
  map(e => e.target.value),
  auditTime(400)
).subscribe(finalColor => {
  applyColor(finalColor);
  addToRecentColors(finalColor);
});

// Practical: Volume slider final value
fromEvent(volumeSlider, 'input').pipe(
  map(e => e.target.value),
  auditTime(200)
).subscribe(finalVolume => {
  setVolume(finalVolume);
  saveVolumePreference(finalVolume);
});

// Practical: Zoom level final value
fromEvent(canvas, 'wheel').pipe(
  map(e => e.deltaY),
  scan((zoom, delta) => {
    const newZoom = zoom + (delta > 0 ? -0.1 : 0.1);
    return Math.max(0.1, Math.min(5, newZoom));
  }, 1),
  auditTime(150)
).subscribe(finalZoom => {
  applyZoom(finalZoom);
  renderAtZoomLevel(finalZoom);
});

import { fromEvent, interval } from 'rxjs';
import { sampleTime, map, scan } from 'rxjs/operators';

// Basic sampleTime - periodic sampling
interval(100).pipe(
  sampleTime(1000)
).subscribe(val => console.log('Sampled:', val));
// Samples every 1 second

// Practical: Mouse position sampling
fromEvent(document, 'mousemove').pipe(
  map(e => ({ x: e.clientX, y: e.clientY })),
  sampleTime(500) // Sample every 500ms
).subscribe(position => {
  logMousePosition(position);
  // Reduces logging from potentially 1000s to 2/second
});

// Practical: Real-time metrics sampling
const cpuUsage$ = interval(50).pipe(
  map(() => getCPUUsage())
);

cpuUsage$.pipe(
  sampleTime(5000) // Sample every 5 seconds
).subscribe(usage => {
  displayMetric('CPU', usage);
  sendToMonitoring(usage);
});

// Practical: Sensor data sampling
const temperatureSensor$ = interval(100).pipe(
  map(() => readTemperature())
);

temperatureSensor$.pipe(
  sampleTime(10000) // Sample every 10 seconds
).subscribe(temp => {
  logTemperature(temp);
  checkThresholds(temp);
});

// Practical: Network bandwidth monitoring
const bytesTransferred$ = new Subject<number>();

bytesTransferred$.pipe(
  scan((total, bytes) => total + bytes, 0),
  sampleTime(1000) // Sample every second
).subscribe(totalBytes => {
  const bandwidth = calculateBandwidth(totalBytes);
  updateBandwidthDisplay(bandwidth);
});

// Practical: Stock price ticker
const priceUpdates$ = websocket$.pipe(
  filter(msg => msg.type === 'price'),
  map(msg => msg.price)
);

priceUpdates$.pipe(
  sampleTime(1000) // Update display every second
).subscribe(price => {
  updatePriceTicker(price);
  // Prevents excessive UI updates
});

// Practical: Audio level meter
const audioLevel$ = new Subject<number>();

audioLevel$.pipe(
  sampleTime(100) // Sample 10 times per second
).subscribe(level => {
  updateAudioMeter(level);
  checkForClipping(level);
});

// Practical: GPS location sampling
const gpsUpdates$ = new Subject<Location>();

gpsUpdates$.pipe(
  sampleTime(30000) // Sample every 30 seconds
).subscribe(location => {
  saveLocationHistory(location);
  updateMapPosition(location);
});

// Practical: Form field activity monitoring
const formActivity$ = merge(
  fromEvent(input1, 'input'),
  fromEvent(input2, 'input'),
  fromEvent(select1, 'change')
).pipe(
  mapTo(Date.now())
);

formActivity$.pipe(
  sampleTime(60000) // Check every minute
).subscribe(lastActivity => {
  updateSessionTimeout(lastActivity);
});

// Practical: Chart data point collection
const dataStream$ = new Subject<number>();

dataStream$.pipe(
  sampleTime(2000) // Add point every 2 seconds
).subscribe(value => {
  addDataPointToChart(value);
  // Prevents chart from becoming too dense
});

// Practical: Scroll progress indicator
fromEvent(window, 'scroll').pipe(
  map(() => {
    const scrolled = window.scrollY;
    const height = document.documentElement.scrollHeight - window.innerHeight;
    return (scrolled / height) * 100;
  }),
  sampleTime(200)
).subscribe(progress => {
  updateProgressBar(progress);
});

// Practical: Performance monitoring
const performanceMetrics$ = interval(100).pipe(
  map(() => ({
    fps: measureFPS(),
    memory: performance.memory?.usedJSHeapSize,
    timestamp: Date.now()
  }))
);

performanceMetrics$.pipe(
  sampleTime(5000)
).subscribe(metrics => {
  logPerformanceMetrics(metrics);
});

import { fromEvent, interval } from 'rxjs';
import { timeInterval, map, tap, take } from 'rxjs/operators';

// Basic timeInterval - measure spacing
interval(1000).pipe(
  take(5),
  timeInterval()
).subscribe(({ value, interval }) => {
  console.log(`Value: ${value}, Interval: ${interval}ms`);
});
// Output shows ~1000ms intervals

// Practical: Click rate detection
fromEvent(button, 'click').pipe(
  timeInterval(),
  tap(({ interval }) => {
    if (interval < 200) {
      console.log('Rapid clicking detected!');
    }
  })
).subscribe(({ value, interval }) => {
  console.log(`Clicked (${interval}ms since last)`);
});

// Practical: API response time monitoring
ajax.getJSON('/api/data').pipe(
  timeInterval()
).subscribe(({ value, interval }) => {
  console.log(`Response time: ${interval}ms`);
  metrics.recordResponseTime(interval);
  
  if (interval > 2000) {
    console.warn('Slow API response');
  }
});

// Practical: User typing speed analysis
fromEvent(input, 'input').pipe(
  timeInterval(),
  map(({ interval }) => interval),
  scan((acc, interval) => {
    acc.intervals.push(interval);
    acc.average = acc.intervals.reduce((a, b) => a + b) / acc.intervals.length;
    return acc;
  }, { intervals: [], average: 0 })
).subscribe(stats => {
  console.log(`Avg typing interval: ${stats.average.toFixed(0)}ms`);
  displayTypingSpeed(stats);
});

// Practical: Network event spacing
const networkEvents$ = merge(
  fromEvent(connection, 'message').pipe(mapTo('message')),
  fromEvent(connection, 'error').pipe(mapTo('error')),
  fromEvent(connection, 'close').pipe(mapTo('close'))
);

networkEvents$.pipe(
  timeInterval()
).subscribe(({ value, interval }) => {
  console.log(`${value} after ${interval}ms`);
  
  if (value === 'error' && interval < 1000) {
    console.error('Rapid errors - connection unstable');
  }
});

// Practical: Frame rate measurement
const frames$ = interval(0, animationFrameScheduler).pipe(
  take(120)
);

frames$.pipe(
  timeInterval(),
  scan((acc, { interval }) => {
    acc.frameTime = interval;
    acc.fps = 1000 / interval;
    return acc;
  }, { frameTime: 0, fps: 0 })
).subscribe(stats => {
  displayFPS(stats.fps);
  
  if (stats.fps < 30) {
    console.warn('Low frame rate detected');
  }
});

// Practical: Message rate limiting detection
const messages$ = new Subject<Message>();

messages$.pipe(
  timeInterval(),
  filter(({ interval }) => interval < 100),
  scan((count) => count + 1, 0)
).subscribe(rapidCount => {
  if (rapidCount > 10) {
    console.warn('Message spam detected');
    throttleMessages();
  }
});

// Practical: Keyboard event analysis
fromEvent(document, 'keydown').pipe(
  timeInterval(),
  map(({ value, interval }) => ({
    key: value.key,
    interval
  }))
).subscribe(({ key, interval }) => {
  console.log(`Key '${key}' pressed ${interval}ms after previous`);
  
  if (interval < 50) {
    console.log('Rapid key press (possible key repeat)');
  }
});

// Practical: Stream health monitoring
dataStream$.pipe(
  timeInterval(),
  tap(({ interval }) => {
    if (interval > 5000) {
      console.warn('Stream stalled - no data for 5s');
      checkStreamHealth();
    }
  })
).subscribe();

// Practical: Request pacing analysis
requestStream$.pipe(
  timeInterval(),
  scan((stats, { interval }) => {
    stats.count++;
    stats.totalTime += interval;
    stats.avgInterval = stats.totalTime / stats.count;
    stats.minInterval = Math.min(stats.minInterval, interval);
    stats.maxInterval = Math.max(stats.maxInterval, interval);
    return stats;
  }, {
    count: 0,
    totalTime: 0,
    avgInterval: 0,
    minInterval: Infinity,
    maxInterval: 0
  })
).subscribe(stats => {
  displayRequestStats(stats);
});

// Practical: Heartbeat monitoring
heartbeat$.pipe(
  timeInterval(),
  tap(({ interval }) => {
    if (interval > 35000) { // Expected every 30s
      console.error('Missed heartbeat');
      attemptReconnect();
    }
  })
).subscribe();

import { fromEvent, interval } from 'rxjs';
import { timestamp, map, scan, take } from 'rxjs/operators';

// Basic timestamp - add absolute timestamp
interval(1000).pipe(
  take(3),
  timestamp()
).subscribe(({ value, timestamp }) => {
  const date = new Date(timestamp);
  console.log(`Value: ${value} at ${date.toISOString()}`);
});

// Practical: Click event logging
fromEvent(button, 'click').pipe(
  timestamp(),
  map(({ value, timestamp }) => ({
    event: 'click',
    element: value.target.id,
    timestamp,
    date: new Date(timestamp).toISOString()
  }))
).subscribe(log => {
  console.log('Click log:', log);
  sendToAnalytics(log);
});

// Practical: Error logging with timestamps
errors$.pipe(
  timestamp(),
  tap(({ value, timestamp }) => {
    const errorLog = {
      error: value,
      timestamp,
      datetime: new Date(timestamp).toISOString(),
      stackTrace: value.stack
    };
    logError(errorLog);
    sendToErrorTracking(errorLog);
  })
).subscribe();

// Practical: User activity timeline
const userActions$ = merge(
  fromEvent(document, 'click').pipe(mapTo('click')),
  fromEvent(document, 'keypress').pipe(mapTo('keypress')),
  fromEvent(document, 'scroll').pipe(mapTo('scroll'))
);

userActions$.pipe(
  timestamp(),
  scan((timeline, { value, timestamp }) => {
    timeline.push({
      action: value,
      timestamp,
      time: new Date(timestamp).toLocaleTimeString()
    });
    return timeline.slice(-50); // Keep last 50 actions
  }, [])
).subscribe(timeline => {
  updateActivityTimeline(timeline);
});

// Practical: Message timestamps for chat
const messages$ = new Subject<string>();

messages$.pipe(
  timestamp(),
  map(({ value, timestamp }) => ({
    text: value,
    timestamp,
    formattedTime: new Date(timestamp).toLocaleTimeString(),
    sender: currentUser.id
  }))
).subscribe(message => {
  displayMessage(message);
  saveToHistory(message);
});

// Practical: Performance event tracking
const performanceEvents$ = merge(
  fromEvent(window, 'load').pipe(mapTo('page_load')),
  ajaxComplete$.pipe(mapTo('ajax_complete')),
  renderComplete$.pipe(mapTo('render_complete'))
);

performanceEvents$.pipe(
  timestamp(),
  scan((metrics, { value, timestamp }) => {
    metrics[value] = timestamp;
    
    if (metrics.page_load && metrics.render_complete) {
      metrics.timeToInteractive = 
        metrics.render_complete - metrics.page_load;
    }
    
    return metrics;
  }, {})
).subscribe(metrics => {
  reportPerformanceMetrics(metrics);
});

// Practical: Sensor data logging
const sensorData$ = interval(1000).pipe(
  map(() => ({
    temperature: readTemperature(),
    humidity: readHumidity(),
    pressure: readPressure()
  }))
);

sensorData$.pipe(
  timestamp(),
  map(({ value, timestamp }) => ({
    ...value,
    timestamp,
    datetime: new Date(timestamp).toISOString()
  }))
).subscribe(reading => {
  saveSensorReading(reading);
  if (reading.temperature > 30) {
    logAlert({
      type: 'high_temp',
      value: reading.temperature,
      timestamp: reading.timestamp
    });
  }
});

// Practical: API request/response logging
ajax.getJSON('/api/data').pipe(
  timestamp(),
  tap(({ timestamp }) => {
    console.log(`Request completed at: ${new Date(timestamp).toISOString()}`);
  })
).subscribe();

// Practical: Form submission audit trail
fromEvent(form, 'submit').pipe(
  timestamp(),
  map(({ value, timestamp }) => ({
    formData: new FormData(value.target),
    submittedAt: timestamp,
    submittedBy: currentUser.id,
    formattedDate: new Date(timestamp).toISOString()
  }))
).subscribe(submission => {
  auditLog.record(submission);
  processSubmission(submission);
});

// Practical: State change history
stateChanges$.pipe(
  timestamp(),
  scan((history, { value, timestamp }) => {
    history.push({
      state: value,
      timestamp,
      date: new Date(timestamp).toISOString()
    });
    return history.slice(-100); // Keep last 100 changes
  }, [])
).subscribe(history => {
  saveStateHistory(history);
  enableUndoRedo(history);
});

// Practical: WebSocket message timestamps
websocket$.pipe(
  timestamp(),
  map(({ value, timestamp }) => ({
    ...value,
    receivedAt: timestamp,
    latency: timestamp - value.sentAt
  }))
).subscribe(message => {
  console.log(`Message latency: ${message.latency}ms`);
  processMessage(message);
});

// Practical: Download progress timestamps
downloadProgress$.pipe(
  timestamp(),
  map(({ value, timestamp }) => ({
    bytesDownloaded: value,
    timestamp,
    elapsedTime: timestamp - downloadStartTime,
    speed: value / ((timestamp - downloadStartTime) / 1000) // bytes/sec
  }))
).subscribe(progress => {
  updateProgressBar(progress);
  displayDownloadSpeed(progress.speed);
});

// Compare timeInterval vs timestamp
const source$ = interval(1000).pipe(take(3));

// timeInterval: relative time between emissions
source$.pipe(timeInterval())
  .subscribe(x => console.log('Interval:', x));
// { value: 0, interval: 1000 }, { value: 1, interval: 1000 }

// timestamp: absolute time of emission
source$.pipe(timestamp())
  .subscribe(x => console.log('Timestamp:', x));
// { value: 0, timestamp: 1703347200000 }, ...