<commit_before>package com.nytimes.android.external.store3.pipeline

import com.nytimes.android.external.store3.base.impl.MemoryPolicy
import kotlinx.coroutines.ExperimentalCoroutinesApi
import kotlinx.coroutines.FlowPreview
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.flow

@FlowPreview
fun <Key, Output> beginPipeline(
    fetcher: (Key) -> Flow<Output>
): PipelineStore<Key, Output> {
    return PipelineFetcherStore(fetcher)
}

// this really needs a better name :/
@FlowPreview
fun <Key, Output> beginNonFlowingPipeline(
    fetcher: suspend (Key) -> Output
): PipelineStore<Key, Output> {
    return PipelineFetcherStore {
        flow {
            emit(fetcher(it))
        }
    }
}


@FlowPreview
fun <Key, OldOutput, NewOutput> PipelineStore<Key, OldOutput>.withConverter(
    converter: suspend (OldOutput) -> NewOutput
): PipelineStore<Key, NewOutput> {
    return PipelineConverterStore(this) { key, value ->
        converter(value)
    }
}

@FlowPreview
fun <Key, OldOutput, NewOutput> PipelineStore<Key, OldOutput>.withKeyConverter(
    converter: suspend (Key, OldOutput) -> NewOutput
): PipelineStore<Key, NewOutput> {
    return PipelineConverterStore(this, converter)
}

@FlowPreview
fun <Key, Output> PipelineStore<Key, Output>.withCache(
    memoryPolicy: MemoryPolicy? = null
): PipelineStore<Key, Output> {
    return PipelineCacheStore(this, memoryPolicy)
}

@FlowPreview
fun <Key, OldOutput, NewOutput> PipelineStore<Key, OldOutput>.withPersister(
    reader: (Key) -> Flow<NewOutput>,
    writer: suspend (Key, OldOutput) -> Unit,
    delete: (suspend (Key) -> Unit)? = null
): PipelineStore<Key, NewOutput> {
    return PipelinePersister(
        fetcher = this,
        reader = reader,
        writer = writer,
        delete = delete
    )
}

@ExperimentalCoroutinesApi
@FlowPreview
fun <Key, OldOutput, NewOutput> PipelineStore<Key, OldOutput>.withNonFlowPersister(
    reader: suspend (Key) -> NewOutput?,
    writer: suspend (Key, OldOutput) -> Unit,
    delete: (suspend (Key) -> Unit)? = null
): PipelineStore<Key, NewOutput> {
    val flowable = SimplePersisterAsFlowable(
        reader = reader,
        writer = writer,
        delete = delete
    )
    return PipelinePersister(
        fetcher = this,
        reader = flowable::flowReader,
        writer = flowable::flowWriter,
        delete = if (delete == null) {
            null
        } else {
            flowable::flowDelete
        }
    )
}
<commit_msg>let persister flow nullables to understand absense. (#30)<commit_after>package com.nytimes.android.external.store3.pipeline

import com.nytimes.android.external.store3.base.impl.MemoryPolicy
import kotlinx.coroutines.ExperimentalCoroutinesApi
import kotlinx.coroutines.FlowPreview
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.flow

@FlowPreview
fun <Key, Output> beginPipeline(
    fetcher: (Key) -> Flow<Output>
): PipelineStore<Key, Output> {
    return PipelineFetcherStore(fetcher)
}

// this really needs a better name :/
@FlowPreview
fun <Key, Output> beginNonFlowingPipeline(
    fetcher: suspend (Key) -> Output
): PipelineStore<Key, Output> {
    return PipelineFetcherStore {
        flow {
            emit(fetcher(it))
        }
    }
}


@FlowPreview
fun <Key, OldOutput, NewOutput> PipelineStore<Key, OldOutput>.withConverter(
    converter: suspend (OldOutput) -> NewOutput
): PipelineStore<Key, NewOutput> {
    return PipelineConverterStore(this) { key, value ->
        converter(value)
    }
}

@FlowPreview
fun <Key, OldOutput, NewOutput> PipelineStore<Key, OldOutput>.withKeyConverter(
    converter: suspend (Key, OldOutput) -> NewOutput
): PipelineStore<Key, NewOutput> {
    return PipelineConverterStore(this, converter)
}

@FlowPreview
fun <Key, Output> PipelineStore<Key, Output>.withCache(
    memoryPolicy: MemoryPolicy? = null
): PipelineStore<Key, Output> {
    return PipelineCacheStore(this, memoryPolicy)
}

@FlowPreview
fun <Key, OldOutput, NewOutput> PipelineStore<Key, OldOutput>.withPersister(
    reader: (Key) -> Flow<NewOutput?>,
    writer: suspend (Key, OldOutput) -> Unit,
    delete: (suspend (Key) -> Unit)? = null
): PipelineStore<Key, NewOutput> {
    return PipelinePersister(
        fetcher = this,
        reader = reader,
        writer = writer,
        delete = delete
    )
}

@ExperimentalCoroutinesApi
@FlowPreview
fun <Key, OldOutput, NewOutput> PipelineStore<Key, OldOutput>.withNonFlowPersister(
    reader: suspend (Key) -> NewOutput?,
    writer: suspend (Key, OldOutput) -> Unit,
    delete: (suspend (Key) -> Unit)? = null
): PipelineStore<Key, NewOutput> {
    val flowable = SimplePersisterAsFlowable(
        reader = reader,
        writer = writer,
        delete = delete
    )
    return PipelinePersister(
        fetcher = this,
        reader = flowable::flowReader,
        writer = flowable::flowWriter,
        delete = if (delete == null) {
            null
        } else {
            flowable::flowDelete
        }
    )
}

{'fraction_non_alphanumeric': 0.11119984028748253, 'fraction_numerical': 0.0011978438810141745, 'mean_word_length': 3.542157751586582, 'pattern_counts': {'":': 0, '<': 45, '<?xml version=': 0, '>': 67, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5046582', 'n_tokens_mistral': 1533, 'n_tokens_neox': 1535, 'n_words': 355}