<commit_before>{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE GeneralizedNewtypeDeriving #-}
module PubSub where

import Control.Applicative
import Control.Monad.Writer
import qualified Data.ByteString.Char8 as B
import Data.Maybe

import Reply


newtype PubSub a = PubSub (WriterT [PubSubAction] IO a)
    deriving (Monad, MonadIO)

data PubSubAction = PubSubAction B.ByteString [B.ByteString]


data Message = Subscribe B.ByteString Integer
             | Unsubscribe B.ByteString Integer
             | PSubscribe B.ByteString Integer
             | PUnsubscribe B.ByteString Integer
             | Message B.ByteString B.ByteString
             | PMessage B.ByteString B.ByteString B.ByteString
    deriving (Show)


readMsg :: Reply -> Maybe Message
readMsg (MultiBulk (Just (r0:r1:r2:rs))) = do
    kind <- readBulk r0
    case kind of
        "subscribe"    -> Subscribe    <$> readBulk r1 <*> readInt r2
        "unsubscribe"  -> Unsubscribe  <$> readBulk r1 <*> readInt r2
        "psubscribe"   -> PSubscribe   <$> readBulk r1 <*> readInt r2
        "punsubscribe" -> PUnsubscribe <$> readBulk r1 <*> readInt r2
        "message"      -> Message      <$> readBulk r1 <*> readBulk r2
        "pmessage"     -> PMessage     <$> readBulk r1
                                            <*> readBulk r2
                                            <*> (maybeHead rs >>= readBulk)
        _              -> Nothing
readMsg _ = Nothing


maybeHead :: [a] -> Maybe a
maybeHead (x:xs) = Just x
maybeHead _      = Nothing



readBulk :: Reply -> Maybe B.ByteString
readBulk (Bulk s) = s
readBulk _        = Nothing

readInt :: Reply -> Maybe Integer
readInt (Integer i) = Just i
readInt _           = Nothing


subscribe chans = PubSubAction "SUBSCRIBE" chans
unsubscribe = PubSubAction "UNSUBSCRIBE"


--pubSub :: PubSub () -> (Message -> PubSub ()) -> Redis a

--subscribe, unsubscribe :: ByteString -> WriterT [PubSub] Redis

{- 
Die an "pubSub" übergebene Action gibt eine Liste an weiteren 


Für jede Message ein Callback. Callbackfunktion hat den von Redis gewrappten
Typ (z.B. IO). Dazu ein WriterT für weitere PubSub-Commands.

Nach dem Callback werden die zurückgegebenen Commands an den Server geschickt.<commit_msg>rename read* -> from*<commit_after>{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE GeneralizedNewtypeDeriving #-}
module PubSub where

import Control.Applicative
import Control.Monad.Writer
import qualified Data.ByteString.Char8 as B
import Data.Maybe

import Reply


newtype PubSub a = PubSub (WriterT [PubSubAction] IO a)
    deriving (Monad, MonadIO)

data PubSubAction = PubSubAction B.ByteString [B.ByteString]


data Message = Subscribe B.ByteString Integer
             | Unsubscribe B.ByteString Integer
             | PSubscribe B.ByteString Integer
             | PUnsubscribe B.ByteString Integer
             | Message B.ByteString B.ByteString
             | PMessage B.ByteString B.ByteString B.ByteString
    deriving (Show)


readMsg :: Reply -> Maybe Message
readMsg (MultiBulk (Just (r0:r1:r2:rs))) = do
    kind <- fromBulk r0
    case kind of
        "subscribe"    -> Subscribe    <$> fromBulk r1 <*> fromInt r2
        "unsubscribe"  -> Unsubscribe  <$> fromBulk r1 <*> fromInt r2
        "psubscribe"   -> PSubscribe   <$> fromBulk r1 <*> fromInt r2
        "punsubscribe" -> PUnsubscribe <$> fromBulk r1 <*> fromInt r2
        "message"      -> Message      <$> fromBulk r1 <*> fromBulk r2
        "pmessage"     -> PMessage     <$> fromBulk r1
                                            <*> fromBulk r2
                                            <*> (maybeHead rs >>= fromBulk)
        _              -> Nothing
readMsg _ = Nothing


maybeHead :: [a] -> Maybe a
maybeHead (x:xs) = Just x
maybeHead _      = Nothing



fromBulk :: Reply -> Maybe B.ByteString
fromBulk (Bulk s) = s
fromBulk _        = Nothing

fromInt :: Reply -> Maybe Integer
fromInt (Integer i) = Just i
fromInt _           = Nothing


subscribe chans = PubSubAction "SUBSCRIBE" chans
unsubscribe = PubSubAction "UNSUBSCRIBE"


--pubSub :: PubSub () -> (Message -> PubSub ()) -> Redis a

--subscribe, unsubscribe :: ByteString -> WriterT [PubSub] Redis

{- 
Die an "pubSub" übergebene Action gibt eine Liste an weiteren 


Für jede Message ein Callback. Callbackfunktion hat den von Redis gewrappten
Typ (z.B. IO). Dazu ein WriterT für weitere PubSub-Commands.

Nach dem Callback werden die zurückgegebenen Commands an den Server geschickt.
{'fraction_non_alphanumeric': 0.09502465262214253, 'fraction_numerical': 0.00761990138951143, 'mean_word_length': 2.4785658612626658, 'pattern_counts': {'":': 0, '<': 31, '<?xml version=': 0, '>': 64, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '152004', 'n_tokens_mistral': 1399, 'n_tokens_neox': 1383, 'n_words': 419}