Finish with the implementation and test

hydra-node/hydra-node.cabal

@@ -139,7 +139,6 @@ library
     , cborg
     , containers
     , contra-tracer
-    , modern-uri
     , data-default
     , directory
     , filepath
@@ -152,7 +151,10 @@ library
     , io-classes
     , iohk-monitoring
     , iproute
+    , lens
+    , lens-aeson
     , memory
+    , modern-uri
     , network
     , network-mux
     , optparse-applicative

hydra-node/src/Hydra/API/Server.hs

@@ -13,27 +13,29 @@ import Hydra.Prelude hiding (TVar, readTVar, seq)
 
 import Control.Concurrent.STM (TChan, dupTChan, readTChan)
 import qualified Control.Concurrent.STM as STM
-import Text.URI
 import Control.Concurrent.STM.TChan (newBroadcastTChanIO, writeTChan)
 import Control.Concurrent.STM.TVar (TVar, modifyTVar', newTVarIO, readTVar)
 import Control.Exception (IOException)
 import qualified Data.Aeson as Aeson
 import Hydra.API.ClientInput (ClientInput)
-import Hydra.API.ServerOutput (ServerOutput (Greetings, InvalidInput), TimedServerOutput (..), replaceTxToCBOR)
+import Hydra.API.ServerOutput (ServerOutput (Greetings, InvalidInput), TimedServerOutput (..), prepareServerOutputResponse)
 import Hydra.Chain (IsChainState)
 import Hydra.Logging (Tracer, traceWith)
 import Hydra.Network (IP, PortNumber)
 import Hydra.Party (Party)
 import Hydra.Persistence (PersistenceIncremental (..))
 import Network.WebSockets (
+  PendingConnection (pendingRequest),
+  RequestHead (..),
   acceptRequest,
   receiveData,
   runServer,
   sendTextData,
   sendTextDatas,
-  withPingThread, RequestHead (..), PendingConnection (pendingRequest)
+  withPingThread,
  )
 import Test.QuickCheck (oneof)
+import Text.URI
 
 data APIServerLog
   = APIServerStarted {listeningPort :: PortNumber}
@@ -169,7 +171,7 @@ runAPIServer host port party tracer history callback responseChannel = do
 
   sendOutputs chan con displayCBORTx = forever $ do
     response <- STM.atomically $ readTChan chan
-    let sentResponse = prepareResponse displayCBORTx response
+    let sentResponse = prepareServerOutputResponse displayCBORTx response
 
     sendTextData con sentResponse
     traceWith tracer (APIOutputSent $ toJSON response)
@@ -194,12 +196,6 @@ runAPIServer host port party tracer history callback responseChannel = do
     hist <- STM.atomically (readTVar history)
     let encodeAndReverse xs serverOutput = Aeson.encode serverOutput : xs
     sendTextDatas con $ foldl' encodeAndReverse [] hist
-  prepareResponse displayCBORTx response =
-    let encodedResponse = Aeson.encode response
-    in
-      if displayCBORTx
-        then replaceTxToCBOR response encodedResponse
-        else encodedResponse
 
 data RunServerException = RunServerException
   { ioException :: IOException

hydra-node/src/Hydra/API/ServerOutput.hs

@@ -2,14 +2,19 @@
 
 module Hydra.API.ServerOutput where
 
-import Data.Aeson (Value (..), withObject, (.:))
+import Cardano.Binary (toStrictByteString)
+import Control.Lens ((?~))
+import Data.Aeson (Value (..), encode, withObject, (.:))
 import qualified Data.Aeson.KeyMap as KeyMap
+import Data.Aeson.Lens (atKey)
+import Data.ByteString.Char8 (unpack)
+import qualified Data.ByteString.Lazy as LBS
+import Data.Text (pack)
 import Hydra.API.ClientInput (ClientInput (..))
 import Hydra.Chain (ChainStateType, HeadId, IsChainState, PostChainTx, PostTxError)
 import Hydra.Crypto (MultiSignature)
 import Hydra.Ledger (IsTx, UTxOType, ValidationError)
 import Hydra.Network (NodeId)
-import qualified Data.ByteString.Lazy as LBS
 import Hydra.Party (Party)
 import Hydra.Prelude hiding (seq)
 import Hydra.Snapshot (Snapshot, SnapshotNumber)
@@ -117,35 +122,42 @@ instance
     PostTxOnChainFailed p e -> PostTxOnChainFailed <$> shrink p <*> shrink e
     RolledBack -> []
 
--- grab response json and just replace tx field encoded as CBOR
-replaceTxToCBOR :: TimedServerOutput tx -> LBS.ByteString -> LBS.ByteString
-replaceTxToCBOR timedOutput encodedResponse =
-  case output timedOutput of
-    PeerConnected {} -> encodedResponse
-    PeerDisconnected {} -> encodedResponse
-    HeadIsInitializing {} -> encodedResponse
-    Committed {} -> encodedResponse
-    HeadIsOpen {} -> encodedResponse
-    HeadIsClosed {} -> encodedResponse
-    HeadIsContested {} -> encodedResponse
-    ReadyToFanout {} -> encodedResponse
-    HeadIsAborted {} -> encodedResponse
-    HeadIsFinalized {} -> encodedResponse
-    CommandFailed {clientInput} ->
+-- | Replaces json encoded tx field to its cbor representation
+-- NOTE: we deliberately pattern match on all 'ServerOutput' constructors
+-- so that we don't forget to update this function if they change.
+prepareServerOutputResponse :: IsChainState tx => Bool -> TimedServerOutput tx -> LBS.ByteString
+prepareServerOutputResponse False response = encode response
+prepareServerOutputResponse True response =
+  case output response of
+    PeerConnected{} -> encodedResponse
+    PeerDisconnected{} -> encodedResponse
+    HeadIsInitializing{} -> encodedResponse
+    Committed{} -> encodedResponse
+    HeadIsOpen{} -> encodedResponse
+    HeadIsClosed{} -> encodedResponse
+    HeadIsContested{} -> encodedResponse
+    ReadyToFanout{} -> encodedResponse
+    HeadIsAborted{} -> encodedResponse
+    HeadIsFinalized{} -> encodedResponse
+    CommandFailed{clientInput} ->
       case clientInput of
         Init -> encodedResponse
         Abort -> encodedResponse
-        Commit {} -> encodedResponse
-        NewTx {Hydra.API.ClientInput.transaction} -> undefined
+        Commit{} -> encodedResponse
+        NewTx{Hydra.API.ClientInput.transaction = tx} -> replacedResponse tx
         GetUTxO -> encodedResponse
         Close -> encodedResponse
         Contest -> encodedResponse
         Fanout -> encodedResponse
-    TxValid {Hydra.API.ServerOutput.transaction} -> undefined
-    TxInvalid {Hydra.API.ServerOutput.transaction} -> undefined
-    SnapshotConfirmed {} -> encodedResponse
-    GetUTxOResponse {} -> encodedResponse
-    InvalidInput {} -> encodedResponse
-    Greetings {} -> encodedResponse
-    PostTxOnChainFailed {} -> encodedResponse
+    TxValid{Hydra.API.ServerOutput.transaction = tx} -> replacedResponse tx
+    TxInvalid{Hydra.API.ServerOutput.transaction = tx} -> replacedResponse tx
+    SnapshotConfirmed{} -> encodedResponse
+    GetUTxOResponse{} -> encodedResponse
+    InvalidInput{} -> encodedResponse
+    Greetings{} -> encodedResponse
+    PostTxOnChainFailed{} -> encodedResponse
     RolledBack -> encodedResponse
+ where
+  encodedResponse = encode response
+  replacedResponse tx =
+    encodedResponse & atKey "transaction" ?~ String (pack . unpack $ toStrictByteString $ toCBOR tx)

hydra-node/test/Hydra/API/ServerSpec.hs

@@ -2,10 +2,12 @@
 
 module Hydra.API.ServerSpec where
 
-import Hydra.Prelude hiding (seq)
+import Hydra.Prelude hiding (decodeUtf8, seq)
 import Test.Hydra.Prelude
 
+import Codec.CBOR.Write (toStrictByteString)
 import Control.Exception (IOException)
+import Control.Lens ((^?))
 import Control.Monad.Class.MonadSTM (
   check,
   modifyTVar',
@@ -16,8 +18,13 @@ import Control.Monad.Class.MonadSTM (
   writeTQueue,
  )
 import qualified Data.Aeson as Aeson
+import Data.Aeson.Lens (key, nonNull)
+import Data.ByteString.Char8 (unpack)
+import qualified Data.List as List
+import Data.Text (pack)
 import Hydra.API.Server (Server (Server, sendOutput), withAPIServer)
-import Hydra.API.ServerOutput (ServerOutput (Greetings, InvalidInput, RolledBack), TimedServerOutput (..), input)
+import Hydra.API.ServerOutput (ServerOutput (..), TimedServerOutput (..), input)
+import Hydra.Chain (HeadId (HeadId))
 import Hydra.Ledger.Simple (SimpleTx)
 import Hydra.Logging (nullTracer, showLogsOnFailure)
 import Hydra.Persistence (PersistenceIncremental (..), createPersistenceIncremental)
@@ -128,17 +135,44 @@ spec = parallel $ do
                   (output <$> timedOutputs) `shouldBe` [greeting]
 
               -- send another message from server
-              sendOutput RolledBack
-              -- Receive one more message and expect it to be 'RolledBack'.
+              newOutput :: ServerOutput SimpleTx <- generate arbitrary
+              sendOutput newOutput
+
+              -- Receive one more message and expect it to be the same as sent one.
               -- This means other messages we sent after the 'Greeting' are ignored as expected.
               received <- replicateM 1 (receiveData conn)
 
               case traverse Aeson.eitherDecode received of
                 Left{} -> failure $ "Failed to decode messages:\n" <> show received
                 Right timedOutputs' -> do
-                  (output <$> timedOutputs') `shouldBe` [RolledBack :: ServerOutput SimpleTx]
+                  (output <$> timedOutputs') `shouldBe` [newOutput]
               return ()
 
+  it "outputs tx as cbor if client says so" $
+    monadicIO $ do
+      tx :: SimpleTx <- pick arbitrary
+      run . failAfter 5 $ do
+        withFreePort $ \port ->
+          withAPIServer @SimpleTx "127.0.0.1" (fromIntegral port) alice mockPersistence nullTracer noop $ \Server{sendOutput} -> do
+            let txValidMessage = TxValid{headId = HeadId "some-head-id", transaction = tx}
+            -- client is able to specify they want tx output to be encoded as CBOR
+            withClient port (defaultPath <> "?tx-output=cbor") $ \conn -> do
+              sendOutput txValidMessage
+              -- receive greetings + one more message
+              received :: [ByteString] <- replicateM 2 (receiveData conn)
+              -- make sure tx output is valid tx cbor
+              (received List.!! 1) ^? key "transaction" . nonNull
+                `shouldBe` Just (Aeson.String . pack . unpack . toStrictByteString $ toCBOR tx)
+
+            -- if client doesn't specify anything they will get tx encoded as JSON
+            withClient port defaultPath $ \conn -> do
+              sendOutput txValidMessage
+              -- receive greetings + one more message
+              received :: [ByteString] <- replicateM 2 (receiveData conn)
+              -- make sure tx output is valid tx cbor
+              (received List.!! 1) ^? key "transaction" . nonNull
+                `shouldBe` Just (toJSON tx)
+
   it "sequence numbers are continuous and strictly monotonically increasing" $
     monadicIO $ do
       outputs :: [ServerOutput SimpleTx] <- pick arbitrary