1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292
|
//
// This file is part of pluggable-discovery-protocol-handler.
//
// Copyright 2021 ARDUINO SA (http://www.arduino.cc/)
//
// This software is released under the GNU General Public License version 3,
// which covers the main part of arduino-cli.
// The terms of this license can be found at:
// https://www.gnu.org/licenses/gpl-3.0.en.html
//
// You can be released from the requirements of the above licenses by purchasing
// a commercial license. Buying such a license is mandatory if you want to modify or
// otherwise use the software for commercial activities involving the Arduino
// software without disclosing the source code of your own applications. To purchase
// a commercial license, send an email to license@arduino.cc.
//
// Package discovery is a library for handling the Arduino Pluggable-Discovery protocol
// (https://github.com/arduino/tooling-rfcs/blob/main/RFCs/0002-pluggable-discovery.md#pluggable-discovery-api-via-stdinstdout)
//
// The library implements the state machine and the parsing logic to implement a pluggable-discovery client and server.
//
// While implementing a server, all the commands issued by the client are conveniently translated into function calls, in particular
// the methods of the Discovery interface are the only functions that must be implemented to get a fully working pluggable discovery
// using this library.
//
// A usage example is provided in the dummy-discovery package.
package discovery
import (
"bufio"
"encoding/json"
"fmt"
"io"
"regexp"
"strconv"
"strings"
"sync"
)
// Discovery is an interface that represents the business logic that
// a pluggable discovery must implement. The communication protocol
// is completely hidden and it's handled by a DiscoveryServer.
type Discovery interface {
// Hello is called once at startup to provide the userAgent string
// and the protocolVersion negotiated with the client.
Hello(userAgent string, protocolVersion int) error
// StartSync is called to put the discovery in event mode. When the
// function returns the discovery must send port events ("add" or "remove")
// using the eventCB function.
StartSync(eventCB EventCallback, errorCB ErrorCallback) error
// Stop stops the discovery internal subroutines. If the discovery is
// in event mode it must stop sending events through the eventCB previously
// set.
Stop() error
// Quit is called just before the server terminates. This function can be
// used by the discovery as a last chance gracefully close resources.
Quit()
}
// EventCallback is a callback function to call to transmit port
// metadata when the discovery is in "sync" mode and a new event
// is detected.
type EventCallback func(event string, port *Port)
// ErrorCallback is a callback function to signal unrecoverable errors to the
// client while the discovery is in event mode. Once the discovery signal an
// error it means that no more port-events will be delivered until the client
// performs a STOP+START_SYNC cycle.
type ErrorCallback func(err string)
// A Server is a pluggable discovery protocol handler,
// it must be created using the NewServer function.
type Server struct {
impl Discovery
userAgent string
reqProtocolVersion int
initialized bool
started bool
syncStarted bool
cachedPorts map[string]*Port
cachedErr string
output io.Writer
outputMutex sync.Mutex
}
// NewServer creates a new discovery server backed by the
// provided pluggable discovery implementation. To start the server
// use the Run method.
func NewServer(impl Discovery) *Server {
return &Server{
impl: impl,
}
}
// Run starts the protocol handling loop on the given input and
// output stream, usually `os.Stdin` and `os.Stdout` are used.
// The function blocks until the `QUIT` command is received or
// the input stream is closed. In case of IO error the error is
// returned.
func (d *Server) Run(in io.Reader, out io.Writer) error {
d.output = out
reader := bufio.NewReader(in)
for {
fullCmd, err := reader.ReadString('\n')
if err != nil {
d.send(messageError("command_error", err.Error()))
return err
}
fullCmd = strings.TrimSpace(fullCmd)
split := strings.Split(fullCmd, " ")
cmd := strings.ToUpper(split[0])
if !d.initialized && cmd != "HELLO" && cmd != "QUIT" {
d.send(messageError("command_error", fmt.Sprintf("First command must be HELLO, but got '%s'", cmd)))
continue
}
switch cmd {
case "HELLO":
if len(fullCmd) < 7 {
d.hello("")
} else {
d.hello(fullCmd[6:])
}
case "START":
d.start()
case "LIST":
d.list()
case "START_SYNC":
d.startSync()
case "STOP":
d.stop()
case "QUIT":
d.impl.Quit()
d.send(messageOk("quit"))
return nil
default:
d.send(messageError("command_error", fmt.Sprintf("Command %s not supported", cmd)))
}
}
}
func (d *Server) hello(cmd string) {
if d.initialized {
d.send(messageError("hello", "HELLO already called"))
return
}
re := regexp.MustCompile(`^(\d+) "([^"]+)"$`)
matches := re.FindStringSubmatch(cmd)
if len(matches) != 3 {
d.send(messageError("hello", "Invalid HELLO command"))
return
}
d.userAgent = matches[2]
v, err := strconv.ParseInt(matches[1], 10, 64)
if err != nil {
d.send(messageError("hello", "Invalid protocol version: "+matches[2]))
return
}
d.reqProtocolVersion = int(v)
if err := d.impl.Hello(d.userAgent, 1); err != nil {
d.send(messageError("hello", err.Error()))
return
}
d.send(&message{
EventType: "hello",
ProtocolVersion: 1, // Protocol version 1 is the only supported for now...
Message: "OK",
})
d.initialized = true
}
func (d *Server) start() {
if d.started {
d.send(messageError("start", "Discovery already STARTed"))
return
}
if d.syncStarted {
d.send(messageError("start", "Discovery already START_SYNCed, cannot START"))
return
}
d.cachedPorts = map[string]*Port{}
d.cachedErr = ""
if err := d.impl.StartSync(d.eventCallback, d.errorCallback); err != nil {
d.send(messageError("start", "Cannot START: "+err.Error()))
return
}
d.started = true
d.send(messageOk("start"))
}
func (d *Server) eventCallback(event string, port *Port) {
id := port.Address + "|" + port.Protocol
if event == "add" {
d.cachedPorts[id] = port
}
if event == "remove" {
delete(d.cachedPorts, id)
}
}
func (d *Server) errorCallback(msg string) {
d.cachedErr = msg
}
func (d *Server) list() {
if !d.started {
d.send(messageError("list", "Discovery not STARTed"))
return
}
if d.syncStarted {
d.send(messageError("list", "discovery already START_SYNCed, LIST not allowed"))
return
}
if d.cachedErr != "" {
d.send(messageError("list", d.cachedErr))
return
}
ports := []*Port{}
for _, port := range d.cachedPorts {
ports = append(ports, port)
}
d.send(&message{
EventType: "list",
Ports: &ports,
})
}
func (d *Server) startSync() {
if d.syncStarted {
d.send(messageError("start_sync", "Discovery already START_SYNCed"))
return
}
if d.started {
d.send(messageError("start_sync", "Discovery already STARTed, cannot START_SYNC"))
return
}
if err := d.impl.StartSync(d.syncEvent, d.errorEvent); err != nil {
d.send(messageError("start_sync", "Cannot START_SYNC: "+err.Error()))
return
}
d.syncStarted = true
d.send(messageOk("start_sync"))
}
func (d *Server) stop() {
if !d.syncStarted && !d.started {
d.send(messageError("stop", "Discovery already STOPped"))
return
}
if err := d.impl.Stop(); err != nil {
d.send(messageError("stop", "Cannot STOP: "+err.Error()))
return
}
d.started = false
if d.syncStarted {
d.syncStarted = false
}
d.send(messageOk("stop"))
}
func (d *Server) syncEvent(event string, port *Port) {
d.send(&message{
EventType: event,
Port: port,
})
}
func (d *Server) errorEvent(msg string) {
d.send(messageError("start_sync", msg))
}
func (d *Server) send(msg *message) {
data, err := json.MarshalIndent(msg, "", " ")
if err != nil {
// We are certain that this will be marshalled correctly
// so we don't handle the error
data, _ = json.MarshalIndent(messageError("command_error", err.Error()), "", " ")
}
data = append(data, '\n')
d.outputMutex.Lock()
defer d.outputMutex.Unlock()
n, err := d.output.Write(data)
if n != len(data) || err != nil {
panic("ERROR")
}
}
|