How to use the xstream.Stream.create function in xstream

// Generic types can't be spread
      // https://github.com/Microsoft/TypeScript/issues/16780
      // https://github.com/Microsoft/TypeScript/issues/10727
      const { store, ...props } = this.props as any
      this.store = store as Store<s>
      this.componentProps = props as P

      const hasStore = Boolean(this.store)

      this.sources = {
        props: Stream.createWithMemory(),
        state: Stream.createWithMemory(),
        store: hasStore ? this.store.getStoreStream() : Stream.never(),
        lifecycle: {
          componentDidMount: Stream.create(),
          componentWillUnmount: Stream.create(),
          componentWillMount: Stream.create(),
          componentWillReceiveProps: Stream.create(),
          shouldComponentUpdate: Stream.create(),
          componentWillUpdate: Stream.create(),
          componentDidUpdate: Stream.create(),
        },
      }

      this.sinks = mainFn(this.sources)
      const {
        stateReducer = [],
        storeReducer = [],
        sideEffect = [],
        initialState = {} as L
      } = this.sinks
</s>

): SimpleMessagingConnection {
  const producer: Producer = {
    start: listener => {
      // tslint:disable-next-line:no-object-mutation
      worker.onmessage = event => {
        listener.next(parseJsonRpcResponse(event.data));
      };
    },
    stop: () => {
      // tslint:disable-next-line:no-object-mutation
      worker.onmessage = null;
    },
  };

  return {
    responseStream: Stream.create(producer),
    sendRequest: request => worker.postMessage(request),
  };
}

if (closeEvent.wasClean) {
            this.eventProducerListener.complete();
          } else {
            this.eventProducerListener.error("Socket was closed unclean");
          }
        }
      },
      timeout,
    );
    this.connected = this.socket.connected;

    const eventProducer: Producer = {
      start: listener => (this.eventProducerListener = listener),
      stop: () => (this.eventProducerListener = undefined),
    };
    this.events = Stream.create(eventProducer);
  }

}
            }
          };

          await poll();
          pollIntervalLogs = setInterval(poll, this.pollIntervalMs);
        },
        stop: () => {
          if (pollIntervalLogs) {
            clearInterval(pollIntervalLogs);
            pollIntervalLogs = undefined;
          }
        },
      };

      const mergedStream = Stream.merge(Stream.create(fromScraperProducer), Stream.create(fromLogsProducer));
      const deduplicatedStream = mergedStream.compose(dropDuplicates(ct => ct.transactionId));
      return deduplicatedStream;
    } else {
      throw new Error("Unsupported query.");
    }
  }

listener.error(error);
          }
        };

        pollInternal = setInterval(poll, defaultPollInterval);
        await poll();
      },
      stop: () => {
        if (pollInternal) {
          clearInterval(pollInternal);
          pollInternal = undefined;
        }
      },
    };

    return Stream.create(producer);
  }

public constructor(url: string, timeout = 10_000, reconnectedHandler?: () => void) {
    this.url = url;
    this.timeout = timeout;
    this.reconnectedHandler = reconnectedHandler;

    const eventProducer: Producer = {
      start: listener => (this.eventProducerListener = listener),
      stop: () => (this.eventProducerListener = undefined),
    };
    this.events = Stream.create(eventProducer);
    this.connectionStatusProducer = new DefaultValueProducer(ConnectionStatus.Unconnected);
    this.connectionStatus = new ValueAndUpdates(this.connectionStatusProducer);

    this.socket = new StreamingSocket(this.url, this.timeout);
    this.socket.events.subscribe({
      next: event => {
        if (!this.eventProducerListener) throw new Error("No event producer listener set");
        this.eventProducerListener.next(event);
      },
      error: () => this.connectionStatusProducer.update(ConnectionStatus.Disconnected),
    });
  }

if (pollInternal) {
              clearTimeout(pollInternal);
              pollInternal = undefined;
            }
            listener.error(error);
          }
        }, defaultPollInterval);
      },
      stop: () => {
        if (pollInternal) {
          clearTimeout(pollInternal);
          pollInternal = undefined;
        }
      },
    };
    return Stream.create(producer);
  }

public constructor(url: string, timeout = 10_000, reconnectedHandler?: () => void) {
    const eventProducer: Producer = {
      start: listener => (this.eventProducerListener = listener),
      stop: () => (this.eventProducerListener = undefined),
    };
    this.events = Stream.create(eventProducer);

    this.socket = new QueueingStreamingSocket(url, timeout, reconnectedHandler);
    this.socket.events.subscribe({
      next: event => {
        if (this.eventProducerListener) {
          this.eventProducerListener.next(event);
        }
      },
      error: error => {
        if (this.eventProducerListener) {
          this.eventProducerListener.error(error);
        }
      },
    });

    this.connectionStatus = this.socket.connectionStatus;

}
            },
            error: error => {
              listener.error(error);
              reset();
            },
          }),
        );
      });
    },
    stop: () => {
      reset();
    },
  };

  return Stream.create(producer);
}

export function fromEvent(
  element: Element | Document,
  eventName: string,
  useCapture = false,
  preventDefault: PreventDefaultOpt = false,
  passive = false
): Stream {
  let next: ((e: Event) => void) | null = null;
  return Stream.create({
    start: function start(listener: Listener) {
      if (preventDefault) {
        next = function _next(event: Event) {
          preventDefaultConditional(event, preventDefault);
          listener.next(event);
        };
      } else {
        next = function _next(event: Event) {
          listener.next(event);
        };
      }
      element.addEventListener(eventName, next, {
        capture: useCapture,
        passive,
      });
    },