How to use the @nteract/messaging.outputs function in @nteract/messaging

// All actions for updating cell status
    cellMessages.pipe(
      kernelStatuses(),
      map(status => actions.updateCellStatus({ id, status, contentRef }))
    ),

    // Update the input numbering: `[ ]`
    cellMessages.pipe(
      executionCounts(),
      map(ct => actions.updateCellExecutionCount({ id, value: ct, contentRef }))
    ),

    // All actions for new outputs
    cellMessages.pipe(
      outputs(),
      map(output => actions.appendOutput({ id, output, contentRef }))
    ),

    // clear_output display message
    cellMessages.pipe(
      ofMessageType("clear_output"),
      mapTo(actions.clearOutputs({ id, contentRef }))
    )
  );

  // On subscription, send the message
  return Observable.create(observer => {
    const subscription = cellAction$.subscribe(observer);
    channels.next(executeRequest);
    return subscription;
  });

map((status: string) =>
        actions.updateCellStatus({ id, status, contentRef })
      )
    ),

    // Update the input numbering: `[ ]`
    cellMessages.pipe(
      executionCounts() as any,
      map((ct: number) =>
        actions.updateCellExecutionCount({ id, value: ct, contentRef })
      )
    ),

    // All actions for new outputs
    cellMessages.pipe(
      outputs() as any,
      map((output: OnDiskOutput) =>
        actions.appendOutput({ id, output, contentRef })
      )
    ),

    // clear_output display message
    cellMessages.pipe(
      ofMessageType("clear_output") as any,
      mapTo(actions.clearOutputs({ id, contentRef }))
    ),

    // Prompt the user for input
    cellMessages.pipe(
      inputRequests() as any,
      map((inputRequest: InputRequestMessage) => {
        return actions.promptInputRequest({

map((status: string) =>
        actions.updateCellStatus({ id, status, contentRef })
      )
    ),

    // Update the input numbering: `[ ]`
    cellMessages.pipe(
      executionCounts() as any,
      map((ct: number) =>
        actions.updateCellExecutionCount({ id, value: ct, contentRef })
      )
    ),

    // All actions for new outputs
    cellMessages.pipe(
      outputs() as any,
      map((output: OnDiskOutput) =>
        actions.appendOutput({ id, output, contentRef })
      )
    ),

    // clear_output display message
    cellMessages.pipe(
      ofMessageType("clear_output") as any,
      mapTo(actions.clearOutputs({ id, contentRef }))
    ),

    // Prompt the user for input
    cellMessages.pipe(
      inputRequests() as any,
      map((inputRequest: InputRequestMessage) => {
        return actions.promptInputRequest({

send(
    data: any,
    callbacks: any,
    metadata?: any,
    buffers?: ArrayBuffer[] | ArrayBufferView[]
  ): string {
    const message = createCommMessage(
      this.comm_id,
      data,
      this.flattenBufferArrays(buffers)
    );

    const callbackAction$ = this.kernel.channels.pipe(
      childOf(message),
      outputs() as any,
      map((output: any) => this.actions.appendOutput(output))
    );

    Observable.create((observer: Observer) => {
      const subscription = callbackAction$.subscribe(observer);
      this.kernel.channels.next(message);
      return subscription;
    }).subscribe(console.log, console.error, console.log);

    return message.header.msg_id;
  }