Conduit Attoparsec의 입력을 수동으로 종료

Question

syslog 로그 파일을 처리하고 있으며, 각 행을 개별 syslog 항목으로 처리하고 Attoparsec 파서를 사용하여 해당 항목을 구문 분석합니다. 그래서 나는 syslog 항목의 스트림을 생성하기 위해

fileToBS :: IO Handle -> C.Source (ResourceT IO) BS.ByteString 
fileToBS handleMaker = source C.$= bsSplitterConduit 
    where source = CB.sourceIOHandle handleMaker 
     bsSplitterConduit = CB.lines 

을 사용하고있다. 사용하고 있습니다.

parseToLogData:: C.Conduit BS.ByteString (ResourceT IO) (Either CATT.ParseError (CATT.PositionRange, LogData)) 
parseToLogData = CATT.conduitParserEither syslogParser 

syslog 값으로 변환합니다. 시스템 로그 값이 파서에서 (내 자신의 몇 가지 유형의 동의어로) 생성됩니다

syslogParser :: Parser (Priority, Maybe UTCTime, IPAddress, BS.ByteString) 
syslogParser = do 
    pri <- priority <?> "priority parse error" 
    mbDate <- date <?> "date parse error" 
    space 
    srcAddr <- ip 
    space 
    msg <- ATT.takeByteString 
    return LogData{pri = pri, timestamp = mbDate, source = srcAddr, message = "msg"} 

priority :: Parser Priority 
priority = do 
    string "<" 
    digitsString <- takeWhile1 digit 
    string ">" 
    return (RawPriority digitsString) 

date :: Parser (Maybe UTCTime) 
date = do 
    rawDate <- ATT.take 15 
    let stringDate = BS.unpack rawDate 
    let parsedDate = parseTime defaultTimeLocale syslogDateFormat stringDate 
    return parsedDate 

ip :: Parser IPAddress 
ip = do 
    oct0 <- takeWhile1 digit 
    period 
    oct1 <- takeWhile1 digit 
    period 
    oct2 <- takeWhile1 digit 
    period 
    oct3 <- takeWhile1 digit 
    return (oct0, oct1, oct2, oct3) 
--ip = takeWhile1 (\x -> digit x || x == 46) 

space = string " " 
colon = string ":" 
period = string "." 

digit test = (test >= 48 && test <= 57) 
octet = digit 

문제는 시스템 로그 항목 (msg <- ATT.takeByteString)의 나머지는 모두 소요되는 라인입니다. 이 함수는 재개 가능한 파서 (conduit의 attoparsec 라이브러리가 사용하는 것)를 사용하는 경우 종료 신호가 필요하기 때문에 스트림에서 훌륭하게 재생되지 않습니다.

이 동작을 수정하기 위해 빈 바이트 체크를 시도했지만 예상대로 작동하지 않습니다 (https://hackage.haskell.org/package/attoparsec-0.12.1.2/docs/Data-Attoparsec-ByteString.html의 증분 입력 참조). 하나의 구문 분석 된 값으로 syslog 입력 파일 전체를 사용합니다. 이것은 80MB 테스트 파일이므로 초기 필드 추출 후 모든 후속 syslog 메시지를 syslog 값의 메시지 필드에 넣습니다.

여기 "원자 메시지"동작을 시도하고 신호를 보내려는 터미네이터 도관이 있습니다. 나는 왜 그것이 작동하지 않는지 모르겠다.

terminator :: C.Conduit BS.ByteString (ResourceT IO) BS.ByteString 
terminator = C.awaitForever yieldAndAddTerminator 
    where 
    yieldAndAddTerminator bs = do 
     C.yield bs 
     C.yield terminator 
    terminator = "" 

관문 세계에서 UDP 메시지를 데이터의 원자 조각으로 처리하려면 어떻게해야합니까?

이 코드베이스의 사본은 여기 https://github.com/tureus/safe-forwarder에서 찾을 수 있습니다.

Answer 1

아마도 parseToLogData에는 새로운 줄 (ASCII 코드 10)을 사용하지 못하게하는 기능이 있습니다.

takeWhileCE (/= 10) =$= parseToLogData 
dropWhileCE (/= 10) >> dropCE 1 -- flush the rest of it 

당신은 또한 line 콤비 기능을 조사 할 수 있습니다 : 도관 - 콤비 용어, 같은 것을 사용.