나는 중첩 iteratees 올바른 방법이라고 생각하지만,이 경우는 대부분의 일반적인 예로부터는 약간 다른 할 몇 가지 고유 한 문제가있다.
덩어리와 그룹
첫 번째 문제는 데이터 소스 권리를 얻는 것입니다. 기본적으로 논리적 인 구분은 [[ByteString]]에 해당하는 스트림을 제공합니다. 이를 직접 생성하는 열거자를 만드는 경우 스트림의 각 요소는 전체 그룹의 청크가됩니다. 이는 메모리상의 이유로 피하기를 원할 것입니다. 모든 것을 단일화하여 [ByteString]으로 만들 수는 있지만 경계를 다시 도입해야 할 필요가 있습니다. DB가 사용자를 대신하여 작업을 수행하기 때문에 꽤 낭비 적입니다.
지금 그룹의 흐름을 무시하면 데이터를 청크로 나눠야하는 것으로 보입니다. Data.Iteratee.group와
enumGroup :: Enumerator ByteString IO a
enumGroup = enumFromCallback cb()
where
cb() = do
(code, data) <- getResultData
case code of
OPERATION_SUCCEEDED -> return $ Right ((True,()), data)
NO_MORE_DATA -> return $ Right ((False,()), data)
GET_DATA_FAILED -> return $ Left MyException
덩어리는 고정 된 크기 때문에, 당신은 쉽게 할 수있는 덩어리이 : 나는대로이 모델 것입니다.
enumGroupChunked :: Iteratee [ByteString] IO a -> IO (Iteratee ByteString IO a)
enumGroupChunked = enumGroup . joinI . group groupSize
그래서
enumGroupChunked은 기본적으로 스트림 유형을 변경하는 화려한 열거입니다
type Enumerator s m a = Iteratee s m a -> m (Iteratee s m a)
Enumerator이의 유형을 비교. 즉, [ByteString] iteratee 소비자를 사용하고 일반 바이트 체크를 사용하는 iteratee를 반환합니다. 열거 자의 반환 유형은 종종 중요하지 않습니다. 그것은 단순히 출력에서 얻을 수 run (또는 tryRun)로 평가 iteratee, 그래서 여기에 동일한 할 수있는 :
evalGroupChunked :: Iteratee [ByteString] IO a -> IO a
evalGroupChunked i = enumGroupChunked i >>= run
각 그룹, 가장 쉬운 장소에 할 수있는 더 복잡한 처리가있는 경우 그렇게하면 enumGroupChunked 함수에있게됩니다. 그룹
의
스트림 지금이 길에서, 그룹의 흐름에 대해 무엇을해야하는 것입니다? 응답은 어떻게 소비할지에 달려 있습니다. 당신은 본질적으로 독립적으로 스트림의 각 그룹을 치료하려면,이 비슷한 일을 할 것입니다 :
foldStream :: Iteratee [ByteString] IO a -> (b -> a -> b) -> b -> IO b
foldStream iter f acc0 = do
val <- evalGroupChunked iter
res <- getNextItem
case res of
OPERATION_SUCCEEDED -> foldStream iter f $! f acc0 val
NO_MORE_DATA -> return $ f acc0 val
GET_DATA_FAILED -> error "had a problem"
을하지만, 이제 당신은 전체 데이터 세트의 스트림 프로세싱, 단지 개인의 그룹이 아닌 일종의을하고 싶은 말은하자 . 즉, 전체 데이터 집합에 대해 실행할
bigProc :: Iteratee [ByteString] IO a
입니다. 이것은 열거 자의 리턴 iteratee가 유용한 곳이다. 일부 이전 코드는 지금 약간 다를 수 :
enumGroupChunked' :: Iteratee [ByteString] IO a
-> IO (Iteratee ByteString IO (Iteratee [ByteString] IO a))
enumGroupChunked' = enumGroup . group groupSize
procStream :: Iteratee [ByteString] IO a -> a
procStream iter = do
i' <- enumGroupChunked' iter >>= run
res <- getNextItem
case res of
OPERATION_SUCCEEDED -> procStream i'
NO_MORE_DATA -> run i'
GET_DATA_FAILED -> error "had a problem"
중첩 iteratees (즉, Iteratee s1 m (Iteratee s2 m a))이 사용은 약간 드문 일이지만, 여러 열거 자에서 프로세스 데이터를 순차적으로 할 때 특히 유용합니다. 핵심은 바깥 쪽 iteratee가 더 많은 데이터를받을 준비가되어있는 iteratee를 줄 것이라고 run이 인식하는 것입니다. 이 경우 제대로 작동하는 모델입니다. 각 그룹을 독립적으로 열거 할 수 있지만 단일 스트림으로 처리 할 수 있기 때문입니다.
하나의주의 : 내부 iteratee는 어떤 상태로 남아있을 것이다. 그룹의 마지막 청크가 전체 청크보다 작을 수 있다고 가정한다.
는 이는이 경우에 무슨 일이 일어날 지 group의 크기는 1024 덩어리로 데이터를 결합하기 때문에,이 그룹 B의 첫 번째 512 바이트와 함께 그룹 A의 마지막 덩어리를 결합 할 것이다 하지
Group A Group B Group C
1024, 1024, 512 1024, 1024, 1024 1024, 1024, 1024
(joinI)을 종료하기 때문에 코드 예에서 foldStream을 사용하여 문제를 해결할 수 있습니다. 그것은 그룹이 진정으로 독립적이라는 것을 의미합니다. 따라서 당신은 그룹을 그렇게 대우해야합니다. 그룹을 procStream처럼 결합하려면 전체 스트림을 생각해야합니다. 이것이 귀하의 경우라면, 단지 group보다 정교한 것을 사용해야 할 것입니다. 하지 IterIO을 언급, 두 패키지의 장점의 논의에 들어가기없이 Data.Enumerator
대
Data.Iteratee (나는 틀림없이 편견이야), 내가 생각 무엇을 지적하고 싶습니다 둘 사이의 가장 중요한 차이 : 스트림의 추상화.
Data.Iteratee에서 소비자 Iteratee ByteString m a은 임의의 길이의 개념적 ByteString에서 작동하며 한 번에 ByteString의 단일 청크에 액세스 할 수 있습니다.
Data.Enumerator에서 소비자 Iteratee ByteString m a은 개념적 [ByteString]에서 작동하며 한 번에 하나 이상의 요소 (바이트)에 액세스 할 수 있습니다.
이 가장 Data.Iteratee 작업이 Data.Enumerator 작업이 청크 초점을 맞춘 ByteString에서 작동하는 반면, 그들은 하나의 Word8에서 작동합니다 Iteratee ByteString 함께 그 요소에 초점을 맞춘 것을 의미한다.
Data.Iteratee.Iteratee [s] m a === Data.Enumerator.Iteratee s m a입니다.
Data.Iteratee 또는 Data.Enumerator는 실제로 사용한 적이 없지만 그럼에도 불구하고 다음 질문은 관련이 있다고 느낍니다. 청크는 고정 크기입니까? 그리고, 당신은'getResultData'로 부분 덩어리를 가져 왔습니까? 아니면 항상 전체 덩어리로 나옵니까? "그룹"에도 동일한 질문이 적용됩니다. –
그룹의 크기는 고정되어 있지만 그룹에서 청크를 읽는 명령이 한 번의 호출로 그룹 전체를 가져 오는 것은 아닙니다. 바깥 쪽 상자는 안쪽 상자가 하나의 그룹이며 getResultData를 한 번 호출 할 때 읽지 않을 수도 있습니다. – ExternalReality