Repository: ktr0731/evans
Pull request #417 (2021-05-16 21:03:21)

Feature | Rerun Last RPC
Implementation for #416 

- Handled server streaming and unary RPC rerun. Client streaming and bidirectional not supported.
- **Slight change from spec:** `call -r` takes the RPC name and hence has the ability to rerun the last request for each different RPC. Earlier, the scope was limited to just the last call. Now it has been extended to the last call per RPC giving us more flexibility.
- Fix minor typos.

**Tested cases**
| Case | Expected |
| --- | --- |
| Old RPC  message was correct and server is running | Should rerun the last command successfully |
| Old RPC message was correct and server is not running | Should fail at `dial()` and run once the server restarts |
| Old RPC message was incorrect / impartial and server is running  | No previous call is recorded if request fails validation |
| Old RPC message was incorrect / impartial and server is not running | No previous call is recorded if request fails validation |

---

**Testing Scenarios (Unary)**
```sh
 ______
 |  ____|
 | |__    __   __   __ _   _ __    ___
 |  __|   \ \ / /  / _. | | '_ \  / __|
 | |____   \ V /  | (_| | | | | | \__ \
 |______|   \_/    \__,_| |_| |_| |___/

 more expressive universal gRPC client


127.0.0.1:5435> package protos

protos@127.0.0.1:5435> service HelloWorld

protos.HelloWorld@127.0.0.1:5435> call SayHello 
message (TYPE_STRING) => Hi there
{
  "message": "Hi there"
}

protos.HelloWorld@127.0.0.1:5435> call --repeat
command call: argument required

protos.HelloWorld@127.0.0.1:5435> call --repeat SayHello
{
  "message": "Hi there"
}

protos.HelloWorld@127.0.0.1:5435> call --repeat SayHello --enrich
content-type: application/grpc

{
  "message": "Hi there"
}

code: OK
number: 0
message: ""

protos.HelloWorld@127.0.0.1:5435> call SayHello
message (TYPE_STRING) => newer
{
  "message": "newer"
}

protos.HelloWorld@127.0.0.1:5435> call SayHello -r
{
  "message": "newer"
}
```

---

**Testing Scenarios (Server Streaming where `num` controls the number of server responses)**
```sh
mirror.MirrorService@127.0.0.1:5435> call MirrorStream
num (TYPE_INT32) => 3
✔ name
name (TYPE_STRING) => Aman
<repeated> f32 (TYPE_FIXED32) => 2
<repeated> f32 (TYPE_FIXED32) => 
what_time::seconds (TYPE_INT64) => 100
what_time::nanos (TYPE_INT32) => 22
{
  "f32": [
    2
  ],
  "name": "Aman",
  "waitTime": 2,
  "whatTime": "1970-01-01T00:01:40.000000022Z"
}
{
  "f32": [
    2
  ],
  "name": "Aman",
  "waitTime": 2,
  "whatTime": "1970-01-01T00:01:40.000000022Z"
}
{
  "f32": [
    2
  ],
  "name": "Aman",
  "waitTime": 2,
  "whatTime": "1970-01-01T00:01:40.000000022Z"
}

mirror.MirrorService@127.0.0.1:5435> call MirrorStream -r --enrich
content-type: application/grpc
{
  "f32": [
    2
  ],
  "name": "Aman",
  "waitTime": 2,
  "whatTime": "1970-01-01T00:01:40.000000022Z"
}
{
  "f32": [
    2
  ],
  "name": "Aman",
  "waitTime": 2,
  "whatTime": "1970-01-01T00:01:40.000000022Z"
}
{
  "f32": [
    2
  ],
  "name": "Aman",
  "waitTime": 2,
  "whatTime": "1970-01-01T00:01:40.000000022Z"
}
rpc-service: mirror

code: OK
number: 0
message: ""

mirror.MirrorService@127.0.0.1:5435> 
```




ktr0731: I'm sorry for the late reply. I'll review it tomorrow, so could you wait until tomorrow? 🙇 
isopropylcyanide: Thank you for taking out the time to review this, on a weekday :) I've addressed your comments. Let me know if there's anything else that I may have missed. 
codecov[bot]: # [Codecov](https://codecov.io/gh/ktr0731/evans/pull/417?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=ktr) Report
> Merging [#417](https://codecov.io/gh/ktr0731/evans/pull/417?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=ktr) (197962d) into [master](https://codecov.io/gh/ktr0731/evans/commit/869a21506bbb8c6ce7e97d9b8f7719616e436f66?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=ktr) (869a215) will **decrease** coverage by `0.25%`.
> The diff coverage is `72.09%`.


```diff
@@            Coverage Diff             @@
##           master     #417      +/-   ##
==========================================
- Coverage   80.39%   80.14%   -0.26%     
==========================================
  Files          57       57              
  Lines        2760     2790      +30     
==========================================
+ Hits         2219     2236      +17     
- Misses        310      320      +10     
- Partials      231      234       +3     
```

isopropylcyanide: Thanks. Will work on the comments this week.

Patch:
diff --git a/.gitignore b/.gitignore
index 8f088f3..39493e7 100644
--- a/.gitignore
+++ b/.gitignore
@@ -8,6 +8,7 @@
 # Folders
 _obj
 _test
+.idea/
 
 # Architecture specific extensions/prefixes
 *.[568vq]
diff --git a/app/commands.go b/app/commands.go
index 82406f3..e61a052 100644
--- a/app/commands.go
+++ b/app/commands.go
@@ -109,7 +109,7 @@ func newOldCommand(flags *flags, ui cui.UI) *command {
 				}
 			}()
 
-			isCLIMode := (cfg.cli || mode.IsCLIMode(cfg.file))
+			isCLIMode := cfg.cli || mode.IsCLIMode(cfg.file)
 			if cfg.repl || !isCLIMode {
 				return runREPLCommand(cfg, ui)
 			}
@@ -280,8 +280,8 @@ func runREPLCommand(cfg *mergedConfig, ui cui.UI) error {
 		return errors.Wrap(err, "failed to run REPL mode")
 	}
 
-	// Always call cancel func because it is hope to abort update checking if REPL mode is finished
-	// before update checking. If update checking is finished before REPL mode, cancel do nothing.
+	// Always call cancel() because it is hope to abort update checking if REPL mode is finished
+	// before update checking. If the update check is finished before REPL mode, cancel does nothing.
 	cancel()
 	if err := eg.Wait(); err != nil {
 		return errors.Wrap(err, "failed to check application update")
diff --git a/cache/cache.go b/cache/cache.go
index b55f5da..453fdc8 100644
--- a/cache/cache.go
+++ b/cache/cache.go
@@ -68,7 +68,7 @@ var Get = func() (*Cache, error) { // Use variable for mocking from means_dev.go
 			return nil, errors.Wrap(err, "failed to create a new cache file")
 		}
 	} else if err != nil {
-		return nil, errors.Wrapf(err, "failed to check the existency of cache file '%s'", p)
+		return nil, errors.Wrapf(err, "failed to check the existence of cache file '%s'", p)
 	}
 
 	f, err := os.Open(p)
diff --git a/repl/commands.go b/repl/commands.go
index 1fa773c..fa403f3 100644
--- a/repl/commands.go
+++ b/repl/commands.go
@@ -155,7 +155,7 @@ func (c *showCommand) Run(w io.Writer, args []string) error {
 }
 
 type callCommand struct {
-	enrich, digManually, bytesFromFile, emitDefaults bool
+	enrich, digManually, bytesFromFile, emitDefaults, repeatCall bool
 }
 
 func (c *callCommand) FlagSet() (*pflag.FlagSet, bool) {
@@ -165,6 +165,7 @@ func (c *callCommand) FlagSet() (*pflag.FlagSet, bool) {
 	fs.BoolVar(&c.digManually, "dig-manually", false, "prompt asks whether to dig down if it encountered to a message field")
 	fs.BoolVar(&c.bytesFromFile, "bytes-from-file", false, "interpret TYPE_BYTES input as a relative path to a file")
 	fs.BoolVar(&c.emitDefaults, "emit-defaults", false, "render fields with default values")
+	fs.BoolVarP(&c.repeatCall, "repeat", "r", false, "repeat previous unary or server streaming request (if exists)")
 	return fs, true
 }
 
@@ -199,7 +200,7 @@ func (c *callCommand) Run(w io.Writer, args []string) error {
 
 	// here we create the request context
 	// we also add the call command flags here
-	err := usecase.CallRPCInteractively(context.Background(), w, args[0], c.digManually, c.bytesFromFile)
+	err := usecase.CallRPCInteractively(context.Background(), w, args[0], c.digManually, c.bytesFromFile, c.repeatCall)
 	if errors.Is(err, io.EOF) {
 		return errors.New("inputting canceled")
 	}
diff --git a/repl/completer.go b/repl/completer.go
index 2cd0190..943ef2d 100644
--- a/repl/completer.go
+++ b/repl/completer.go
@@ -81,8 +81,8 @@ func newCompleter(cmds map[string]commander) *completer {
 					s = []*prompt.Suggest{
 						prompt.NewSuggestion("package", "show loaded package names"),
 						prompt.NewSuggestion("service", "show loaded service names"),
-						prompt.NewSuggestion("message", "show loaded messsage names"),
-						prompt.NewSuggestion("rpc", "show RPC names belong to the current selected service"),
+						prompt.NewSuggestion("message", "show loaded message names"),
+						prompt.NewSuggestion("rpc", "show RPC names belonging to the current selected service"),
 						prompt.NewSuggestion("header", "show headers which will be added to each request"),
 					}
 				}
diff --git a/usecase/call_rpc.go b/usecase/call_rpc.go
index efa8341..4103aff 100644
--- a/usecase/call_rpc.go
+++ b/usecase/call_rpc.go
@@ -7,8 +7,11 @@ import (
 	"sync"
 	"time"
 
-	"github.com/ktr0731/evans/fill"
+	"github.com/jhump/protoreflect/dynamic"
+	"github.com/ktr0731/evans/grpc"
 	"github.com/ktr0731/evans/idl/proto"
+
+	"github.com/ktr0731/evans/fill"
 	"github.com/ktr0731/evans/logger"
 	"github.com/pkg/errors"
 	"golang.org/x/sync/errgroup"
@@ -46,21 +49,33 @@ func (e *gRPCError) Code() ErrorCode {
 // the request to the gRPC server and decodes the response body to res.
 // Note that req and res must be JSON-decodable structs. The output is written to w.
 func CallRPC(ctx context.Context, w io.Writer, rpcName string) error {
-	return dm.CallRPC(ctx, w, rpcName, dm.filler)
+	return dm.CallRPC(ctx, w, rpcName, false, dm.filler)
 }
-func (m *dependencyManager) CallRPC(ctx context.Context, w io.Writer, rpcName string, filler fill.Filler) error {
+func (m *dependencyManager) CallRPC(ctx context.Context, w io.Writer, rpcName string, rerunPrevious bool, filler fill.Filler) error {
 	fqsn := proto.FullyQualifiedServiceName(m.state.selectedPackage, m.state.selectedService)
 	rpc, err := m.spec.RPC(fqsn, rpcName)
 	if err != nil {
-		return errors.Wrap(err, "failed to get the RPC descriptor")
+		return errors.Wrapf(err, "failed to get the RPC descriptor for: %s", rpcName)
+	}
+	if rerunPrevious && rpc.IsClientStreaming {
+		return errors.New("cannot rerun previous RPC as client/bidi streaming RPCs are not supported")
 	}
 	newRequest := func() (interface{}, error) {
 		req := rpc.RequestType.New()
-		err = filler.Fill(req)
-		if errors.Is(err, io.EOF) {
-			return nil, io.EOF
+		if !rerunPrevious {
+			err = filler.Fill(req)
+			if errors.Is(err, io.EOF) {
+				return nil, io.EOF
+			}
+			if err != nil {
+				return nil, err
+			}
+			if err := m.updateRPCCallState(rpcName, req); err != nil {
+				return nil, err
+			}
+			return req, nil
 		}
-		if err != nil {
+		if err = m.getPreviousRPCRequest(rpc, req.(*dynamic.Message)); err != nil {
 			return nil, err
 		}
 		return req, nil
@@ -416,6 +431,45 @@ func (m *dependencyManager) CallRPC(ctx context.Context, w io.Writer, rpcName st
 	}
 }
 
+// Gets a request with the body containing the payload of its previous RPC
+// Only RPCs that are repeatable by definition will have their previous requests returned.
+func (m *dependencyManager) getPreviousRPCRequest(rpc *grpc.RPC, req *dynamic.Message) error {
+	id := rpcIdentifier(rpc.Name)
+	if _, ok := m.state.rpcCallState[id]; !ok {
+		return errors.Errorf("no previous request exists for RPC: %s, please issue a normal request", id)
+	}
+	if rpc.IsClientStreaming {
+		return errors.Errorf("cannot rerun previous RPC: %s as client/bidi streaming RPCs are not supported", id)
+	}
+	previousReqBytes := m.state.rpcCallState[id].requestPayload
+	if previousReqBytes == nil {
+		return errors.Errorf("no previous request body exists for RPC: %s, please issue a normal request", id)
+	}
+	err := req.Unmarshal(previousReqBytes)
+	if err != nil {
+		return errors.Wrapf(err, "error while unmarshalling request for RPC: %s, please run without the --repeat option", id)
+	}
+	return nil
+}
+
+// Updates the last call state for the given RPC. This is done by serializing
+// the request payload and store it into the state buffer indexed by the rpcName
+// The RPC is repeatable only if it is not a client streaming rpc.
+func (m *dependencyManager) updateRPCCallState(rpcName string, req interface{}) error {
+	message := req.(*dynamic.Message)
+	reqBytes, err := message.Marshal()
+	if err != nil {
+		return err
+	}
+	if m.state.rpcCallState == nil {
+		m.state.rpcCallState = make(map[rpcIdentifier]callState)
+	}
+	m.state.rpcCallState[rpcIdentifier(rpcName)] = callState{
+		requestPayload: reqBytes,
+	}
+	return nil
+}
+
 type interactiveFiller struct {
 	fillFunc func(v interface{}) error
 }
@@ -424,12 +478,12 @@ func (f *interactiveFiller) Fill(v interface{}) error {
 	return f.fillFunc(v)
 }
 
-func CallRPCInteractively(ctx context.Context, w io.Writer, rpcName string, digManually, bytesFromFile bool) error {
-	return dm.CallRPCInteractively(ctx, w, rpcName, digManually, bytesFromFile)
+func CallRPCInteractively(ctx context.Context, w io.Writer, rpcName string, digManually, bytesFromFile bool, rerunPrevious bool) error {
+	return dm.CallRPCInteractively(ctx, w, rpcName, digManually, bytesFromFile, rerunPrevious)
 }
 
-func (m *dependencyManager) CallRPCInteractively(ctx context.Context, w io.Writer, rpcName string, digManually, bytesFromFile bool) error {
-	return m.CallRPC(ctx, w, rpcName, &interactiveFiller{
+func (m *dependencyManager) CallRPCInteractively(ctx context.Context, w io.Writer, rpcName string, digManually, bytesFromFile, rerunPrevious bool) error {
+	return m.CallRPC(ctx, w, rpcName, rerunPrevious, &interactiveFiller{
 		fillFunc: func(v interface{}) error {
 			return m.interactiveFiller.Fill(v, fill.InteractiveFillerOpts{DigManually: digManually, BytesFromFile: bytesFromFile})
 		},
diff --git a/usecase/usecase.go b/usecase/usecase.go
index 8fa954d..c8a6039 100644
--- a/usecase/usecase.go
+++ b/usecase/usecase.go
@@ -22,14 +22,20 @@ type dependencyManager struct {
 	gRPCClient        grpc.Client
 	responseFormatter *format.ResponseFormatter
 	resourcePresenter present.Presenter
-
-	state state
+	state             state
 }
 
+type rpcIdentifier string
+
 // state has the domain state modified by each usecase logic. The default value is used as the initial value.
 type state struct {
 	selectedPackage string // TODO: remove in v1.0.0.
 	selectedService string
+	rpcCallState    map[rpcIdentifier]callState
+}
+
+type callState struct {
+	requestPayload []byte
 }
 
 type Dependencies struct {


Test patch:
diff --git a/e2e/e2egen/e2egen.go b/e2e/e2egen/e2egen.go
index 5f97a90..fd80ed2 100644
--- a/e2e/e2egen/e2egen.go
+++ b/e2e/e2egen/e2egen.go
@@ -135,7 +135,7 @@ func generateFile(w io.Writer, src, testCaseName string, input, args []string) e
 		l4 := file.LineStart(fset.Position(nextLine).Line + 4)
 		l5 := file.LineStart(fset.Position(nextLine).Line + 5)
 
-		elts := append(compLit.Elts, &ast.KeyValueExpr{
+		compLit.Elts = append(compLit.Elts, &ast.KeyValueExpr{
 			Key: &ast.BasicLit{
 				ValuePos: l1,
 				Kind:     token.STRING,
@@ -182,7 +182,6 @@ func generateFile(w io.Writer, src, testCaseName string, input, args []string) e
 			},
 		})
 		compLit.Rbrace = l5
-		compLit.Elts = elts
 		return false
 	}, nil)
 
diff --git a/e2e/testdata/fixtures/teste2e_repl-call_--help.golden b/e2e/testdata/fixtures/teste2e_repl-call_--help.golden
index 1486230..a8089f4 100644
--- a/e2e/testdata/fixtures/teste2e_repl-call_--help.golden
+++ b/e2e/testdata/fixtures/teste2e_repl-call_--help.golden
@@ -5,4 +5,5 @@ Options:
       --dig-manually      prompt asks whether to dig down if it encountered to a message field
       --emit-defaults     render fields with default values
       --enrich            enrich response output includes header, message, trailer and status
+  -r, --repeat            repeat previous unary or server streaming request (if exists)
 
diff --git a/usecase/call_rpc_test.go b/usecase/call_rpc_test.go
new file mode 100644
index 0000000..4a0de97
--- /dev/null
+++ b/usecase/call_rpc_test.go
@@ -0,0 +1,55 @@
+package usecase
+
+import (
+	"testing"
+
+	"github.com/ktr0731/evans/grpc"
+
+	"github.com/jhump/protoreflect/dynamic"
+)
+
+func TestGetPreviousRPCRequest(t *testing.T) {
+	cases := map[string]struct {
+		expectedError string
+		rpc           *grpc.RPC
+		rpcCallState  map[rpcIdentifier]callState
+	}{
+		"no previous request exists": {
+			rpc:           getStubRPC(false),
+			expectedError: "no previous request exists for RPC: TestRPC, please issue a normal request",
+		},
+		"previous request is client streaming": {
+			rpcCallState:  map[rpcIdentifier]callState{"TestRPC": {}},
+			rpc:           getStubRPC(true),
+			expectedError: "cannot rerun previous RPC: TestRPC as client/bidi streaming RPCs are not supported",
+		},
+		"previous request bytes are nil": {
+			rpcCallState:  map[rpcIdentifier]callState{"TestRPC": {}},
+			rpc:           getStubRPC(false),
+			expectedError: "no previous request body exists for RPC: TestRPC, please issue a normal request",
+		},
+	}
+	for name, c := range cases {
+		c := c
+		t.Run(name, func(t *testing.T) {
+			d := &dependencyManager{
+				state: state{
+					rpcCallState: c.rpcCallState,
+				},
+			}
+			var req *dynamic.Message
+			err := d.getPreviousRPCRequest(c.rpc, req)
+			if err == nil || err.Error() != c.expectedError {
+				t.Errorf("expected error %s, but got %s", c.expectedError, err)
+			}
+		})
+	}
+}
+
+func getStubRPC(clientStreaming bool) *grpc.RPC {
+	return &grpc.RPC{
+		Name:              "TestRPC",
+		IsServerStreaming: true,
+		IsClientStreaming: clientStreaming,
+	}
+}