useSubscription

Signature

function useSubscription<T = unknown>(
  serverId: string,
  topic: string,
  handler: (envelope: IEnvelope<T>) => void | Promise<void>,
  options?: UseSubscriptionOptions,
): UseSubscriptionReturn

interface UseSubscriptionReturn {
  isSubscribed: boolean;
  unsubscribe: () => void;
  error: Error | null;
}

interface UseSubscriptionOptions {
  enabled?: boolean;     // false = skip subscription until true
  onError?: (error: Error) => void;
  autoCleanup?: boolean;
}

Basic Subscription

import { useSubscription } from 'edge-stream-js-react';

function WorkflowPanel() {
  const [tasks, setTasks] = useState([]);

  const { isSubscribed } = useSubscription(
    'bas',
    'workflow.*',
    (envelope) => {
      setTasks(prev => [...prev, {
        id: envelope.meta.id,
        topic: envelope.meta.topic,
        data: envelope.body,
      }]);
    }
  );

  return (
    <div>
      {!isSubscribed && <span>Connecting...</span>}
      <ul>
        {tasks.map(t => <li key={t.id}>{t.topic}</li>)}
      </ul>
    </div>
  );
}

Conditional Subscription (enabled option)

function ConditionalSubscriber({ userId }: { userId: string | null }) {
  const { isSubscribed } = useSubscription(
    'bas',
    `user.${userId}.*`,
    (envelope) => handleUserEvent(envelope),
    {
      enabled: !!userId,  // only subscribe when userId is set
      onError: (err) => console.error('Subscription failed:', err),
    }
  );

  return <div>{isSubscribed ? 'Subscribed' : 'Waiting for user...'}</div>;
}

Manual Unsubscribe

function PausableStream() {
  const { isSubscribed, unsubscribe } = useSubscription(
    'bas',
    'agent.chat.*',
    handleAgentMessage
  );

  return (
    <div>
      <p>Status: {isSubscribed ? 'Active' : 'Stopped'}</p>
      {isSubscribed && (
        <button onClick={unsubscribe}>Pause stream</button>
      )}
    </div>
  );
}

Error Handling

function RobustSubscription() {
  const { isSubscribed, error } = useSubscription(
    'bas',
    'workflow.*',
    handler,
    {
      onError: (err) => {
        // Called if server not found or subscribe() throws
        errorMonitor.capture(err);
      }
    }
  );

  if (error) {
    return <div>Subscription error: {error.message}</div>;
  }

  return <div>{isSubscribed ? 'Live' : 'Connecting...'}</div>;
}

How Cleanup Works

// Internally, useSubscription does this:
useEffect(() => {
  const subscription = client.subscribe(serverId, topic, handler);
  setIsSubscribed(true);

  return () => {
    subscription.unsubscribe(); // called when component unmounts
    setIsSubscribed(false);
  };
}, [client, serverId, topic, handler]);

// Correct: stable handler with useCallback
function StableSubscriber() {
  const dispatch = useDispatch();

  const handler = useCallback((envelope) => {
    dispatch(workflowEventReceived(envelope.body));
  }, [dispatch]);

  useSubscription('bas', 'workflow.*', handler);
  return <div />;
}