Implement Persistent example 💾 (#40)

examples/README.md

@@ -4,4 +4,6 @@ Those examples are ports of those in [Mu Scala](https://github.com/higherkindnes
 
 * Health check
 * Route guide
-* TODO list
+* Simple TODO list
+* Seed example
+* Integration with Persistent (db access)

examples/seed/seed.proto

@@ -1,7 +1,5 @@
 syntax = "proto3";
 
-import "google/protobuf/empty.proto";
-
 package seed;
 
 message Person { string name = 1; int32 age = 2; }

examples/with-persistent/.gitignore

@@ -0,0 +1,3 @@
+.stack-work/
+stack*.yaml.lock
+*~

examples/with-persistent/README.md

@@ -0,0 +1,18 @@
+# with-persistent
+
+## Execution
+
+Running the server:
+
+```bash
+stack run persistent-server
+```
+
+[comment]: # (Start Copyright)
+# Copyright
+
+Mu is designed and developed by 47 Degrees
+
+Copyright (C) 2019-2020 47 Degrees. <http://47deg.com>
+
+[comment]: # (End Copyright)

examples/with-persistent/Setup.hs

@@ -0,0 +1,2 @@
+import Distribution.Simple
+main = defaultMain

examples/with-persistent/mu-example-with-persistent.cabal

@@ -0,0 +1,29 @@
+name:                mu-example-with-persistent
+version:             0.1.0.0
+-- synopsis:
+-- description:
+homepage:            https://github.com/higherkindness/mu-haskell/examples/with-persistent#readme
+author:              Flavio Corpa
+maintainer:          flavio.corpa@47deg.com
+copyright:           Copyright © 2019-2020 47 Degrees. <http://47deg.com>
+category:            Network
+build-type:          Simple
+cabal-version:       >=1.10
+extra-source-files:  README.md
+
+executable persistent-server
+  hs-source-dirs:      src
+  main-is:             Server.hs
+  other-modules:       Schema
+  default-language:    Haskell2010
+  build-depends:       base >= 4.12 && < 5
+                     , conduit
+                     , monad-logger
+                     , mu-schema
+                     , mu-rpc
+                     , mu-protobuf
+                     , mu-grpc-server
+                     , persistent
+                     , persistent-sqlite
+                     , persistent-template
+                     , text

examples/with-persistent/src/Schema.hs

@@ -0,0 +1,49 @@
+{-# language DataKinds                  #-}
+{-# language DeriveGeneric              #-}
+{-# language DuplicateRecordFields      #-}
+{-# language EmptyDataDecls             #-}
+{-# language FlexibleContexts           #-}
+{-# language FlexibleInstances          #-}
+{-# language GADTs                      #-}
+{-# language GeneralizedNewtypeDeriving #-}
+{-# language MultiParamTypeClasses      #-}
+{-# language OverloadedStrings          #-}
+{-# language PolyKinds                  #-}
+{-# language QuasiQuotes                #-}
+{-# language StandaloneDeriving         #-}
+{-# language TemplateHaskell            #-}
+{-# language TypeFamilies               #-}
+{-# language UndecidableInstances       #-}
+
+module Schema where
+
+import           Data.Int                (Int32, Int64)
+import qualified Data.Text               as T
+import           Database.Persist.Sqlite
+import           Database.Persist.TH
+import           GHC.Generics
+
+import           Mu.Quasi.GRpc
+import           Mu.Schema
+
+grpc "PersistentSchema" id "with-persistent.proto"
+
+newtype PersonRequest = PersonRequest
+  { identifier :: Int64
+  } deriving (Eq, Show, Ord, Generic)
+
+instance HasSchema PersistentSchema "PersonRequest" PersonRequest
+
+share [mkPersist sqlSettings, mkMigrate "migrateAll"] [persistLowerCase|
+Person json
+  name T.Text
+  age  Int32
+  deriving Show
+|]
+
+deriving instance Generic Person
+
+-- Unfortunately we need to write this instance by hand 😔 (for now!)
+instance HasSchema PersistentSchema "Person" (Entity Person) where
+  fromSchema (TRecord (Field (FSchematic (TRecord (Field (FPrimitive pid) :* Nil))) :* Field (FPrimitive name) :* Field (FPrimitive age) :* Nil)) = Entity (PersonKey (SqlBackendKey pid)) (Person name age)
+  toSchema (Entity (PersonKey (SqlBackendKey pid)) (Person name age)) = TRecord $ Field (FSchematic (TRecord (Field (FPrimitive pid) :* Nil))) :* Field (FPrimitive name) :* Field (FPrimitive age) :* Nil

examples/with-persistent/src/Server.hs

@@ -0,0 +1,49 @@
+{-# language FlexibleContexts      #-}
+{-# language OverloadedStrings     #-}
+{-# language PartialTypeSignatures #-}
+{-# language TypeApplications      #-}
+{-# language TypeFamilies          #-}
+
+module Main where
+
+import           Control.Monad.IO.Class  (liftIO)
+import           Control.Monad.Logger
+import           Data.Conduit
+import qualified Data.Text               as T
+import           Database.Persist.Sqlite
+import           Mu.GRpc.Server
+import           Mu.Server
+
+import           Schema
+
+main :: IO ()
+main = do
+  putStrLn "running app with persistent"
+  runStderrLoggingT $
+    withSqliteConn @(LoggingT IO) ":memory:" $ \conn -> do
+      liftIO $ flip runSqlPersistM conn $ runMigration migrateAll
+      liftIO $ runGRpcApp 1234 (server conn)
+
+server :: SqlBackend -> ServerT PersistentService ServerErrorIO _
+server p = Server (getPerson p :<|>: newPerson p :<|>: allPeople p :<|>: H0)
+
+runDb = (liftIO .) . flip runSqlPersistM
+
+getPerson :: SqlBackend -> PersonRequest -> ServerErrorIO (Entity Person)
+getPerson conn (PersonRequest idf) = do
+  r <- runDb conn $ do
+    let pId = PersonKey $ SqlBackendKey idf
+    maybePerson <- get pId
+    pure $ Entity pId <$> maybePerson
+  case r of
+    Just p  -> pure p
+    Nothing -> serverError $ ServerError NotFound "unknown person"
+
+newPerson :: SqlBackend -> Entity Person -> ServerErrorIO PersonRequest
+newPerson conn (Entity _ p@(Person name _)) = runDb conn $ do
+  PersonKey (SqlBackendKey nId) <- insert p
+  pure $ PersonRequest nId
+
+allPeople :: SqlBackend -> ConduitT (Entity Person) Void ServerErrorIO () -> ServerErrorIO ()
+allPeople conn sink = runDb conn $
+  runConduit $ selectSource [] [] .| liftServerConduit sink

examples/with-persistent/with-persistent.proto

@@ -0,0 +1,14 @@
+syntax = "proto3";
+
+import "google/protobuf/empty.proto";
+
+package withpersistent;
+
+message PersonRequest { int64 identifier = 1; }
+message Person { PersonRequest pid = 1; string name = 2; int32 age = 3; }
+
+service PersistentService {
+  rpc getPerson (PersonRequest) returns (Person);
+  rpc newPerson (Person) returns (PersonRequest);
+  rpc allPeople (google.protobuf.Empty) returns (stream Person);
+}

grpc/server/src/Mu/GRpc/Server.hs

@@ -17,6 +17,8 @@ module Mu.GRpc.Server
 , runGRpcAppTLS, TLSSettings
   -- * Convert a 'Server' into a WAI application
 , gRpcApp
+  -- * Raise errors as exceptions in IO
+, raiseErrors, liftServerConduit
 ) where
 
 import           Control.Concurrent.Async
@@ -142,14 +144,20 @@ class GRpcMethodHandler m args r h where
   gRpcMethodHandler :: (forall a. m a -> ServerErrorIO a)
                     -> Proxy args -> Proxy r -> RPC -> h -> ServiceHandler
 
-raiseErrors :: ServerErrorIO a -> IO a
+liftServerConduit
+  :: MonadIO m
+  => ConduitT a b ServerErrorIO r -> ConduitT a b m r
+liftServerConduit = transPipe raiseErrors
+
+raiseErrors :: MonadIO m => ServerErrorIO a -> m a
 raiseErrors h
-  = do h' <- runExceptT h
-       case h' of
-         Right r -> return r
-         Left (ServerError code msg)
-           -> closeEarly $ GRPCStatus (serverErrorToGRpcError code)
-                                      (BS.pack msg)
+  = liftIO $ do
+      h' <- runExceptT h
+      case h' of
+        Right r -> return r
+        Left (ServerError code msg)
+          -> closeEarly $ GRPCStatus (serverErrorToGRpcError code)
+                                     (BS.pack msg)
   where
     serverErrorToGRpcError :: ServerErrorCode -> GRPCStatusCode
     serverErrorToGRpcError Unknown         = UNKNOWN
@@ -170,6 +178,27 @@ instance (ProtoBufTypeRef rref r)
     = unary @_ @() @(ViaProtoBufTypeRef rref r)
             rpc (\_ _ -> ViaProtoBufTypeRef <$> raiseErrors (f h))
 
+instance (ProtoBufTypeRef rref r, MonadIO m)
+         => GRpcMethodHandler m '[ ] ('RetStream rref)
+                              (ConduitT r Void m () -> m ()) where
+  gRpcMethodHandler f _ _ rpc h
+    = serverStream @_ @() @(ViaProtoBufTypeRef rref r) rpc sstream
+    where sstream :: req -> ()
+                  -> IO ((), ServerStream (ViaProtoBufTypeRef rref r) ())
+          sstream _ _ = do
+            -- Variable to connect input and output
+            var <- newEmptyTMVarIO :: IO (TMVar (Maybe r))
+            -- Start executing the handler
+            promise <- async (raiseErrors $ ViaProtoBufTypeRef <$> f (h (toTMVarConduit var)))
+              -- Return the information
+            let readNext _
+                  = do nextOutput <- atomically $ takeTMVar var
+                       case nextOutput of
+                         Just o  -> return $ Just ((), ViaProtoBufTypeRef o)
+                         Nothing -> do cancel promise
+                                       return Nothing
+            return ((), ServerStream readNext)
+
 instance (ProtoBufTypeRef vref v)
          => GRpcMethodHandler m '[ 'ArgSingle vref ] 'RetNothing (v -> m ()) where
   gRpcMethodHandler f _ _ rpc h

stack-nightly.yaml

@@ -12,6 +12,7 @@ packages:
 - examples/route-guide
 - examples/seed
 - examples/todolist
+- examples/with-persistent
 - compendium-client
 
 extra-deps:

stack.yaml

@@ -11,6 +11,7 @@ packages:
 - examples/route-guide
 - examples/seed
 - examples/todolist
+- examples/with-persistent
 - compendium-client
 
 extra-deps: