receiveMessage reads a single message off the connection and processes it, returning an error if no message could be received from the connection. Otherwise, (a message is received successfully), the message is processed and any error is handled appropriately. The return value indicates whether the
()
| 340 | // received from the connection. Otherwise, (a message is received successfully), the message is processed and any |
| 341 | // error is handled appropriately. The return value indicates whether the connection should be closed. |
| 342 | func (h *ConnectionHandler) receiveMessage() (bool, error) { |
| 343 | var endOfMessages bool |
| 344 | // For the time being, we handle panics in this function and treat them the same as errors so that they don't |
| 345 | // forcibly close the connection. Contrast this with the panic handling logic in HandleConnection, where we treat any |
| 346 | // panic as unrecoverable to the connection. As we fill out the implementation, we can revisit this decision and |
| 347 | // rethink our posture over whether panics should terminate a connection. |
| 348 | if HandlePanics { |
| 349 | defer func() { |
| 350 | if r := recover(); r != nil { |
| 351 | stackTrace := string(debug.Stack()) |
| 352 | logrus.Errorf("Listener recovered panic: %v: %s", r, stackTrace) |
| 353 | |
| 354 | eomErr := errors.Errorf("receiveMessage recovered panic: %v: %s", r, stackTrace) |
| 355 | if !endOfMessages && h.waitForSync { |
| 356 | if syncErr := h.discardToSync(); syncErr != nil { |
| 357 | fmt.Println(syncErr.Error()) |
| 358 | } |
| 359 | } |
| 360 | h.endOfMessages(eomErr) |
| 361 | } |
| 362 | }() |
| 363 | } |
| 364 | |
| 365 | msg, err := h.backend.Receive() |
| 366 | if err != nil { |
| 367 | return false, errors.Errorf("error receiving message: %w", err) |
| 368 | } |
| 369 | |
| 370 | if m, ok := msg.(json.Marshaler); ok && logrus.IsLevelEnabled(logrus.DebugLevel) { |
| 371 | msgInfo, err := m.MarshalJSON() |
| 372 | if err != nil { |
| 373 | return false, err |
| 374 | } |
| 375 | logrus.Debugf("Received message: %s", string(msgInfo)) |
| 376 | } else { |
| 377 | logrus.Debugf("Received message: %t", msg) |
| 378 | } |
| 379 | |
| 380 | var stop bool |
| 381 | stop, endOfMessages, err = h.handleMessage(msg) |
| 382 | if err != nil { |
| 383 | if !endOfMessages && h.waitForSync { |
| 384 | if syncErr := h.discardToSync(); syncErr != nil { |
| 385 | fmt.Println(syncErr.Error()) |
| 386 | } |
| 387 | } |
| 388 | h.endOfMessages(err) |
| 389 | } else if endOfMessages { |
| 390 | h.endOfMessages(nil) |
| 391 | } |
| 392 | |
| 393 | return stop, nil |
| 394 | } |
| 395 | |
| 396 | // handleMessages processes the message provided and returns status flags indicating what the connection should do next. |
| 397 | // If the |stop| response parameter is true, it indicates that the connection should be closed by the caller. If the |
no test coverage detected