federationsender/roomserver: don't panic while federation is disabled
MODIFY
federationsender/consumers/roomserver.go
federationsender/consumers/roomserver.go
@@ -94,6 +94,12 @@ func (s *OutputRoomEventConsumer) onMessage(msg *sarama.ConsumerMessage) error {
}
if err := s.processMessage(*output.NewRoomEvent); err != nil {
+ switch err.(type) {
+ case *queue.ErrorFederationDisabled:
+ log.WithField("error", output.Type).Info(
+ err.Error(),
+ )
+ default:
// panic rather than continue with an inconsistent database
log.WithFields(log.Fields{
"event": string(ev.JSON()),
@@ -101,6 +107,7 @@ func (s *OutputRoomEventConsumer) onMessage(msg *sarama.ConsumerMessage) error {
"del": output.NewRoomEvent.RemovesStateEventIDs,
log.ErrorKey: err,
}).Panicf("roomserver output log: write room event failure")
+ }
return nil
}
default:

MODIFY
federationsender/queue/queue.go
federationsender/queue/queue.go
@@ -122,13 +122,23 @@ func (oqs *OutgoingQueues) getQueue(destination gomatrixserverlib.ServerName) *d
return oq
}
+type ErrorFederationDisabled struct {
+ Message string
+}
+
+func (e *ErrorFederationDisabled) Error() string {
+ return e.Message
+}
+
// SendEvent sends an event to the destinations
func (oqs *OutgoingQueues) SendEvent(
ev *gomatrixserverlib.HeaderedEvent, origin gomatrixserverlib.ServerName,
destinations []gomatrixserverlib.ServerName,
) error {
if oqs.disabled {
- return fmt.Errorf("federation is disabled")
+ return &ErrorFederationDisabled{
+ Message: "Federation disabled",
+ }
}
if origin != oqs.origin {
// TODO: Support virtual hosting; gh issue #577.
@@ -190,7 +200,9 @@ func (oqs *OutgoingQueues) SendEDU(
destinations []gomatrixserverlib.ServerName,
) error {
if oqs.disabled {
- return fmt.Errorf("federation is disabled")
+ return &ErrorFederationDisabled{
+ Message: "Federation disabled",
+ }
}
if origin != oqs.origin {
// TODO: Support virtual hosting; gh issue #577.

federationsender/roomserver: don't panic while federation is disabled (#1615)