-- ----------------------------------------------------------------------------
{- |
  Module     : Examples.MapReduce.Sort.Sort
  Copyright  : Copyright (C) 2008 Stefan Schmidt
  License    : MIT

  Maintainer : Stefan Schmidt (stefanschmidt@web.de)
  Stability  : experimental
  Portability: portable
  Version    : 0.1


-}
-- ----------------------------------------------------------------------------

module Examples.MapReduce.Sort.Sort
(
  sortAction
, sortActionMap
, namesList
, sortDemoJob
, createSortDemoFiles
)
where

import           Data.Binary
--import           Holumbus.Common.MRBinary

import           System.Log.Logger

import qualified Holumbus.FileSystem.FileSystem as FS

import qualified Holumbus.Data.KeyMap as KMap
import           Holumbus.MapReduce.Types


localLogger :: String
localLogger = "Examples.MapReduce.Sort.Sort"


-- ----------------------------------------------------------------------------
-- Distributed Sort
-- ----------------------------------------------------------------------------


mapSort :: ActionEnvironment -> () -> () -> (String, String) -> IO [(String, (String, String))]
mapSort _ _ _ v
  = do 
    infoM localLogger "mapSort"
    debugM localLogger $ "input: " ++ show v
    let res = [(snd v, v)]
    debugM localLogger $ "output: " ++ show res
    return res


reduceSort :: ActionEnvironment -> () -> String -> [(String, String)] -> IO (Maybe [(String, String)])
reduceSort _ _ k vs 
  = do
    infoM localLogger "reduce/combine Sort"
    debugM localLogger $ "input: " ++ k ++ " - " ++ show vs
    return (Just vs)


-- ----------------------------------------------------------------------------
-- Actions
-- ----------------------------------------------------------------------------

sortAction
  :: ActionConfiguration 
       ()                                          -- state
       () (String, String)                         -- k1, v1
       String (String, String)                     -- k2, v2
       (String, String)                            -- v3 == v2
       [(String, String)]                          -- v4
sortAction
  = (defaultActionConfiguration "SORT")
        { ac_Map     = Just mapAction
        , ac_Combine = Nothing
        , ac_Reduce  = Just reduceAction
        }
    where
      mapAction 
        = (defaultMapConfiguration mapSort)
      reduceAction
        = (defaultReduceConfiguration reduceSort)
        

sortActionMap :: ActionMap
sortActionMap
  = KMap.insert (readActionConfiguration sortAction) $
    KMap.empty


-- ----------------------------------------------------------------------------
-- DemoJob
-- ----------------------------------------------------------------------------

  
sortDemoJob :: JobInfo
sortDemoJob = 
  createJobInfoFromConfiguration
    sortAction -- action config
    ()                  -- options
    namesList           -- input (Tuples)
    []                  -- input (Files)
    1                   -- number of splitters
    2                   -- number of mappers
    1                   -- number of reducers
    1                   -- number of results
    TOTRawTuple         -- type of the result (file of raw)
    
namesList :: [((),(String, String))]
namesList
  = map (\d -> ((),d)) names
  where
  names =
   [("Abraham","Simpson"),
    ("Apu","Nahasapeemapetilon"),
    ("Waylon","Smithers"),
    ("Barney","Gumble"),
    ("Bart","Simpson"),
    ("Carl","Carlson"),
    ("Homer","Simpson"),
    ("Jacqueline","Bouvier"),
    ("Kirk","van Houten"),
    ("Lenny","Leonard"),
    ("Lionel","Hutz"),
    ("Lisa","Simpson"),
    ("Luann","van Houten"),
    ("Maggie","Simpson"),
    ("Manjula","Nahasapeemapetilon"),
    ("Marge","Simpson"),
    ("Martin","Prince"),
    ("Maude","Flanders"),
    ("Milhouse","van Houten"),
    ("Moe","Syzslak"),
    ("Montgomery","Burns"),
    ("Ned","Flanders"),
    ("Patty","Bouvier"),
    ("Rod","Flanders"),
    ("Selma","Bouvier"),
    ("Seymour","Skinner"),
    ("Todd","Flanders"),
    ("Troy","McClure")]
    

-- ----------------------------------------------------------------------------
-- DemoFiles
-- ----------------------------------------------------------------------------


createSortDemoFiles :: FS.FileSystem -> IO ()
createSortDemoFiles fs
  = do
    -- let c = S.BinaryFile (encode ("foo","a aa aaa b bb bbb"))
    -- let c = S.TextFile "harddisk file"
    let c = encode namesList
    FS.createFile "namesList.txt" c fs