You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@plc4x.apache.org by cd...@apache.org on 2021/03/08 11:08:34 UTC
[plc4x] branch develop updated: - Updated the knx codec to check
expectations for expirations,
even if no message was received - Added a dedicated TimeoutError type to
allow easy detection of timeout errors
This is an automated email from the ASF dual-hosted git repository.
cdutz pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/plc4x.git
The following commit(s) were added to refs/heads/develop by this push:
new 06fbedd - Updated the knx codec to check expectations for expirations, even if no message was received - Added a dedicated TimeoutError type to allow easy detection of timeout errors
06fbedd is described below
commit 06fbedde73111e7c02643de93efce76fbabb0504
Author: Christofer Dutz <ch...@c-ware.de>
AuthorDate: Mon Mar 8 12:08:28 2021 +0100
- Updated the knx codec to check expectations for expirations, even if no message was received
- Added a dedicated TimeoutError type to allow easy detection of timeout errors
---
.../plc4go/knxnetip/KnxNetIpMessageCodec.go | 46 ++++++++++++++--------
plc4go/internal/plc4go/spi/errors/TimeoutError.go | 30 ++++++++++++++
2 files changed, 59 insertions(+), 17 deletions(-)
diff --git a/plc4go/internal/plc4go/knxnetip/KnxNetIpMessageCodec.go b/plc4go/internal/plc4go/knxnetip/KnxNetIpMessageCodec.go
index b11f011..24c1737 100644
--- a/plc4go/internal/plc4go/knxnetip/KnxNetIpMessageCodec.go
+++ b/plc4go/internal/plc4go/knxnetip/KnxNetIpMessageCodec.go
@@ -23,6 +23,7 @@ import (
"fmt"
"github.com/apache/plc4x/plc4go/internal/plc4go/knxnetip/readwrite/model"
"github.com/apache/plc4x/plc4go/internal/plc4go/spi"
+ errors2 "github.com/apache/plc4x/plc4go/internal/plc4go/spi/errors"
"github.com/apache/plc4x/plc4go/internal/plc4go/spi/transports"
"github.com/apache/plc4x/plc4go/internal/plc4go/spi/utils"
log "github.com/sirupsen/logrus"
@@ -108,8 +109,6 @@ func (m *KnxNetIpMessageCodec) SendRequest(message interface{}, acceptsMessage s
return err
}
err = m.Expect(acceptsMessage, handleMessage, handleError, ttl)
- // Throttle the throughput.
- time.Sleep(time.Millisecond * 20)
return err
}
@@ -153,9 +152,28 @@ func (m *KnxNetIpMessageCodec) receive() (interface{}, error) {
}
func work(m *KnxNetIpMessageCodec) {
+ // If this method ever exits, start it again.
+ defer work(m)
// Start an endless loop
- // TODO: Provide some means to terminate this ...
for {
+ // Check for any expired expectations.
+ // (Doing this outside the loop lets us expire expectations even if no input is coming in)
+ now := time.Now()
+ for index, expectation := range m.expectations {
+ // Check if this expectation has expired.
+ if now.After(expectation.expiration) {
+ // Remove this expectation from the list.
+ m.expectations = append(m.expectations[:index], m.expectations[index+1:]...)
+ // Call the error handler.
+ err := expectation.handleError(errors2.NewTimeoutError())
+ if err != nil {
+ log.Errorf("Got an error handling error on expectation: %s", err.Error())
+ }
+ continue
+ }
+ }
+
+ // Check for incoming messages.
message, err := m.receive()
if err != nil {
fmt.Printf("got an error reading from transport %s", err.Error())
@@ -182,7 +200,6 @@ func work(m *KnxNetIpMessageCodec) {
}
// Handle the packet itself
- now := time.Now()
// Give a message interceptor a chance to intercept
if m.messageInterceptor != nil {
m.messageInterceptor(message)
@@ -190,18 +207,6 @@ func work(m *KnxNetIpMessageCodec) {
// Go through all expectations
messageHandled := false
for index, expectation := range m.expectations {
- // Check if this expectation has expired.
- if now.After(expectation.expiration) {
- // Remove this expectation from the list.
- m.expectations = append(m.expectations[:index], m.expectations[index+1:]...)
- // Call the error handler.
- err := expectation.handleError(errors.New("expectation timed out"))
- if err != nil {
- log.Errorf("Got an error handling error on expectation: %s", err.Error())
- }
- continue
- }
-
// Check if the current message matches the expectations
// If it does, let it handle the message.
if accepts := expectation.acceptsMessage(message); accepts {
@@ -214,6 +219,12 @@ func work(m *KnxNetIpMessageCodec) {
} else if (index + 1) < len(m.expectations) {
m.expectations = append(m.expectations[:index], m.expectations[index+1:]...)
}
+ } else {
+ // Pass the error to the error handler.
+ err := expectation.handleError(err)
+ if err != nil {
+ log.Errorf("Got an error handling error on expectation: %s", err.Error())
+ }
}
}
}
@@ -223,7 +234,8 @@ func work(m *KnxNetIpMessageCodec) {
m.defaultIncomingMessageChannel <- message
}
} else {
- // Sleep for 10ms
+ // Sleep for 10ms before checking again, in order to not
+ // consume 100% CPU Power.
time.Sleep(time.Millisecond * 10)
}
}
diff --git a/plc4go/internal/plc4go/spi/errors/TimeoutError.go b/plc4go/internal/plc4go/spi/errors/TimeoutError.go
new file mode 100644
index 0000000..7135201
--- /dev/null
+++ b/plc4go/internal/plc4go/spi/errors/TimeoutError.go
@@ -0,0 +1,30 @@
+//
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+//
+package errors
+
+type TimeoutError struct {
+}
+
+func NewTimeoutError() TimeoutError {
+ return TimeoutError{}
+}
+
+func (t TimeoutError) Error() string {
+ return "got timeout"
+}