Skip to content
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.

Commit 2fc87db

Browse files
cmagliecmaglie
committedOct 5, 2021
Implementation of gRPC Monitor command
1 parent 3eb819c commit 2fc87db

File tree

6 files changed

+146
-742
lines changed

6 files changed

+146
-742
lines changed
 

‎commands/daemon/daemon.go

+63-1
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@ package daemon
1717

1818
import (
1919
"context"
20+
"errors"
2021
"fmt"
2122
"io"
2223

@@ -475,5 +476,66 @@ func (s *ArduinoCoreServerImpl) EnumerateMonitorPortSettings(ctx context.Context
475476

476477
// Monitor FIXMEDOC
477478
func (s *ArduinoCoreServerImpl) Monitor(stream rpc.ArduinoCoreService_MonitorServer) error {
478-
return status.New(codes.Unimplemented, "Not implemented").Err()
479+
// The configuration must be sent on the first message
480+
req, err := stream.Recv()
481+
if err != nil {
482+
return err
483+
}
484+
485+
portProxy, _, err := monitor.Monitor(stream.Context(), req)
486+
if err != nil {
487+
return err
488+
}
489+
ctx, cancel := context.WithCancel(stream.Context())
490+
go func() {
491+
defer cancel()
492+
for {
493+
msg, err := stream.Recv()
494+
if errors.Is(err, io.EOF) {
495+
return
496+
}
497+
if err != nil {
498+
stream.Send(&rpc.MonitorResponse{Error: err.Error()})
499+
return
500+
}
501+
if conf := msg.GetPortConfiguration(); conf != nil {
502+
for _, c := range conf.GetSettings() {
503+
if err := portProxy.Config(c.SettingId, c.Value); err != nil {
504+
stream.Send(&rpc.MonitorResponse{Error: err.Error()})
505+
}
506+
}
507+
}
508+
tx := msg.GetTxData()
509+
for len(tx) > 0 {
510+
n, err := portProxy.Write(tx)
511+
if errors.Is(err, io.EOF) {
512+
return
513+
}
514+
if err != nil {
515+
stream.Send(&rpc.MonitorResponse{Error: err.Error()})
516+
return
517+
}
518+
tx = tx[n:]
519+
}
520+
}
521+
}()
522+
go func() {
523+
defer cancel()
524+
buff := make([]byte, 4096)
525+
for {
526+
n, err := portProxy.Read(buff)
527+
if errors.Is(err, io.EOF) {
528+
return
529+
}
530+
if err != nil {
531+
stream.Send(&rpc.MonitorResponse{Error: err.Error()})
532+
return
533+
}
534+
if err := stream.Send(&rpc.MonitorResponse{RxData: buff[:n]}); err != nil {
535+
return
536+
}
537+
}
538+
}()
539+
<-ctx.Done()
540+
return nil
479541
}

‎commands/daemon/term_example/go.mod

-11
This file was deleted.

‎commands/daemon/term_example/go.sum

-676
This file was deleted.

‎commands/daemon/term_example/main.go

+78-49
Original file line numberDiff line numberDiff line change
@@ -17,20 +17,17 @@ package main
1717

1818
import (
1919
"context"
20+
"errors"
2021
"fmt"
22+
"io"
2123
"log"
2224
"time"
2325

24-
monitor "github.com/arduino/arduino-cli/rpc/cc/arduino/cli/monitor/v1"
26+
"github.com/arduino/arduino-cli/rpc/cc/arduino/cli/commands/v1"
2527
"google.golang.org/grpc"
26-
"google.golang.org/protobuf/types/known/structpb"
2728
)
2829

29-
var (
30-
dataDir string
31-
)
32-
33-
// This program exercise monitor rate limiting functionality.
30+
// This program exercise CLI monitor functionality.
3431

3532
func main() {
3633
conn, err := grpc.Dial("localhost:50051", grpc.WithInsecure(), grpc.WithBlock())
@@ -39,58 +36,90 @@ func main() {
3936
}
4037
defer conn.Close()
4138

42-
// Open a monitor instance
43-
mon := monitor.NewMonitorServiceClient(conn)
44-
stream, err := mon.StreamingOpen(context.Background())
45-
if err != nil {
46-
log.Fatal("Error opening stream:", err)
47-
}
39+
// Create and initialize a CLI instance
40+
cli := commands.NewArduinoCoreServiceClient(conn)
4841

49-
additionalConf, err := structpb.NewStruct(
50-
map[string]interface{}{"OutputRate": float64(1000000)},
51-
)
52-
if err != nil {
53-
log.Fatal("Error creating config:", err)
42+
var instance *commands.Instance
43+
if resp, err := cli.Create(context.Background(), &commands.CreateRequest{}); err != nil {
44+
log.Fatal("Create:", err)
45+
} else {
46+
instance = resp.Instance
5447
}
5548

56-
if err := stream.Send(&monitor.StreamingOpenRequest{
57-
Content: &monitor.StreamingOpenRequest_Config{
58-
Config: &monitor.MonitorConfig{
59-
Type: monitor.MonitorConfig_TARGET_TYPE_NULL,
60-
AdditionalConfig: additionalConf,
61-
RecvRateLimitBuffer: 1024,
62-
},
63-
},
64-
}); err != nil {
65-
log.Fatal("Error opening stream:", err)
49+
if respStream, err := cli.Init(context.Background(), &commands.InitRequest{Instance: instance}); err != nil {
50+
log.Fatal("Init:", err)
51+
} else {
52+
for {
53+
resp, err := respStream.Recv()
54+
if errors.Is(err, io.EOF) {
55+
break
56+
}
57+
if err != nil {
58+
log.Fatal("Init:", err)
59+
}
60+
fmt.Println(resp)
61+
}
6662
}
6763

68-
if err := stream.Send(&monitor.StreamingOpenRequest{
69-
Content: &monitor.StreamingOpenRequest_RecvAcknowledge{
70-
RecvAcknowledge: 5,
71-
},
72-
}); err != nil {
73-
log.Fatal("Error replenishing recv window:", err)
64+
// List boards and take the first available port
65+
var port *commands.Port
66+
if resp, err := cli.BoardList(context.Background(), &commands.BoardListRequest{Instance: instance}); err != nil {
67+
log.Fatal("BoardList:", err)
68+
} else {
69+
ports := resp.GetPorts()
70+
if len(ports) == 0 {
71+
log.Fatal("No port to connect!")
72+
}
73+
port = ports[0].Port
7474
}
75+
fmt.Println("Detected port:", port.Label, port.ProtocolLabel)
7576

76-
for {
77-
r, err := stream.Recv()
78-
if err != nil {
79-
log.Fatal("Error receiving from server:", err)
77+
// Connect to the port monitor
78+
fmt.Println("Connecting to monitor")
79+
ctx, cancel := context.WithCancel(context.Background())
80+
if respStream, err := cli.Monitor(ctx); err != nil {
81+
log.Fatal("Monitor:", err)
82+
} else {
83+
if err := respStream.Send(&commands.MonitorRequest{
84+
Instance: instance,
85+
Port: port,
86+
}); err != nil {
87+
log.Fatal("Monitor send-config:", err)
8088
}
81-
if l := len(r.Data); l > 0 {
82-
fmt.Printf("RECV %d bytes\n", l)
89+
time.Sleep(1 * time.Second)
90+
91+
go func() {
92+
for {
93+
if resp, err := respStream.Recv(); err != nil {
94+
fmt.Println(" RECV:", err)
95+
break
96+
} else {
97+
fmt.Println(" RECV:", resp)
98+
}
99+
}
100+
}()
101+
102+
hello := &commands.MonitorRequest{
103+
TxData: []byte("HELLO!"),
83104
}
84-
if r.Dropped > 0 {
85-
fmt.Printf("DROPPED %d bytes!!!\n", r.Dropped)
105+
fmt.Println("Send:", hello)
106+
if err := respStream.Send(hello); err != nil {
107+
log.Fatal("Monitor send HELLO:", err)
86108
}
87-
if err := stream.Send(&monitor.StreamingOpenRequest{
88-
Content: &monitor.StreamingOpenRequest_RecvAcknowledge{
89-
RecvAcknowledge: 1,
90-
},
91-
}); err != nil {
92-
log.Fatal("Error replenishing recv window:", err)
109+
110+
fmt.Println("Send:", hello)
111+
if err := respStream.Send(hello); err != nil {
112+
log.Fatal("Monitor send HELLO:", err)
113+
}
114+
115+
time.Sleep(5 * time.Second)
116+
117+
fmt.Println("Closing Monitor")
118+
if err := respStream.CloseSend(); err != nil {
119+
log.Fatal("Monitor close send:", err)
93120
}
94-
time.Sleep(5 * time.Millisecond)
121+
time.Sleep(5 * time.Second)
95122
}
123+
cancel()
124+
time.Sleep(5 * time.Second)
96125
}

‎i18n/data/en.po

+1-1
Original file line numberDiff line numberDiff line change
@@ -2978,7 +2978,7 @@ msgstr "no compatible version of %s tools found for the current os"
29782978
msgid "no executable specified"
29792979
msgstr "no executable specified"
29802980

2981-
#: commands/daemon/daemon.go:99
2981+
#: commands/daemon/daemon.go:100
29822982
msgid "no instance specified"
29832983
msgstr "no instance specified"
29842984

‎i18n/rice-box.go

+4-4
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

0 commit comments

Comments
 (0)
Please sign in to comment.