class SchemaKSourceFactory {

  public static SchemaKStream<?> buildSource(
      final KsqlQueryBuilder builder,
      final DataSource dataSource,
      final QueryContext.Stacker contextStacker
  ) {
    final boolean windowed = dataSource.getKsqlTopic().getKeyFormat().isWindowed();
    switch (dataSource.getDataSourceType()) {
      case KSTREAM:
        return windowed
            ? buildWindowedStream(
            builder,
            dataSource,
            contextStacker
        ) : buildStream(
            builder,
            dataSource,
            contextStacker
        );

      case KTABLE:
        return windowed
            ? buildWindowedTable(
            builder,
            dataSource,
            contextStacker
        ) : buildTable(
            builder,
            dataSource,
            contextStacker
        );

      default:
        throw new UnsupportedOperationException("Source type:" + dataSource.getDataSourceType());
    }
  }

  private  SchemaKSourceFactory();

}