Reworked connection to use a single thread for state management.
Also disabled the SetConnection API call.
Stream cleanup.
Removed Unnescessary threads, there is now one thread per connection (handling response stream forwarding), and the existing thread is used to forward the request stream.
Renamed 'streams' to 'request'.
Renamed 'nbFrame' to 'requestFrame'.
Renamed 'sbFrame' to 'responseFrame'.
Changed handling of streaming requests.
Incoming & Outgoing streams are split when a connection becomes ready.
Added playback of non-streaming requests/responses for newly opened streams.
Late stream catchup fix & streaming call detection.
Fixed an issue where old streams were not being caught up with what they missed.
Streaming requests & responses are now detected based on the proto definitions.
Changed where the proto file is specified in the afrouter config (see afrouter/arouter.json for an example).
Fixed mutex copy.
Also tweaked some log statements.
Fixed field tag lint error.
Change-Id: I6e14039c27519d8d2103065258ff4302bc881235
diff --git a/afrouter/afrouter/codec.go b/afrouter/afrouter/codec.go
index 675320a..7ea6ef2 100644
--- a/afrouter/afrouter/codec.go
+++ b/afrouter/afrouter/codec.go
@@ -21,7 +21,6 @@
"github.com/golang/protobuf/proto"
"github.com/opencord/voltha-go/common/log"
"google.golang.org/grpc"
- "sync"
)
func Codec() grpc.Codec {
@@ -36,19 +35,18 @@
parentCodec grpc.Codec
}
-// sbFrame is a frame being "returned" to whomever established the connection
-type sbFrame struct {
+// responseFrame is a frame being "returned" to whomever established the connection
+type responseFrame struct {
payload []byte
router Router
method string
backend *backend
- mutex sync.Mutex
metaKey string
metaVal string
}
-// nbFrame is a frame coming in from whomever established the connection
-type nbFrame struct {
+// requestFrame is a frame coming in from whomever established the connection
+type requestFrame struct {
payload []byte
router Router
backend *backend
@@ -61,9 +59,9 @@
func (cdc *transparentRoutingCodec) Marshal(v interface{}) ([]byte, error) {
switch t := v.(type) {
- case *sbFrame:
+ case *responseFrame:
return t.payload, nil
- case *nbFrame:
+ case *requestFrame:
return t.payload, nil
default:
return cdc.parentCodec.Marshal(v)
@@ -72,17 +70,21 @@
func (cdc *transparentRoutingCodec) Unmarshal(data []byte, v interface{}) error {
switch t := v.(type) {
- case *sbFrame:
+ case *responseFrame:
t.payload = data
// This is where the affinity is established on a northbound response
t.router.ReplyHandler(v)
return nil
- case *nbFrame:
+ case *requestFrame:
t.payload = data
// This is were the afinity value is pulled from the payload
// and the backend selected.
t.backend = t.router.Route(v)
- log.Debugf("Routing returned %v for method %s", t.backend, t.methodInfo.method)
+ name := "<nil>"
+ if t.backend != nil {
+ name = t.backend.name
+ }
+ log.Debugf("Routing returned %s for method %s", name, t.methodInfo.method)
return nil
default:
return cdc.parentCodec.Unmarshal(data, v)