### Question:
ArchivingInputQueueListenerTask extends QueueListenerTask { @Override public boolean processMessage(String address, byte[] buff) { String contents = new String(buff); RealtimeEnvelope envelope = null; CcLocationReportRecord record = null; try { envelope = deserializeMessage(contents); if (envelope == null || envelope.getCcLocationReport() == null) { _log.error("Message discarded, probably corrupted, contents= " + contents); Exception e = new Exception( "deserializeMessage failed, possible corrupted message."); _dao.handleException(contents, e, null); return false; } boolean validEnvelope = validationService.validateRealTimeRecord(envelope); if (validEnvelope) { record = new CcLocationReportRecord(envelope, contents, getZoneOffset()); _ccLocationCache.put(record); } else { long vehicleId = envelope.getCcLocationReport().getVehicle().getVehicleId(); _log.error( "Discarding real time record for vehicle : {} as it does not meet the " + "required database constraints", vehicleId); Exception e = new Exception("Real time record for vehile : " + vehicleId + " failed validation." + "Discarding"); _dao.handleException(contents, e, new Date()); } if (record != null) { emergencyStatusNotificationService.process(record); persistor.persist(record); } if (System.currentTimeMillis() - zoneOffsetWindow > 60 * 60 * 1000) { _zoneOffset = null; zoneOffsetWindow = System.currentTimeMillis(); } } catch (Throwable t) { _log.error("Exception processing contents= " + contents, t); try { Date timeReceived = null; if (envelope != null) timeReceived = new Date(envelope.getTimeReceived()); _dao.handleException(contents, t, timeReceived); } catch (Throwable tt) { _log.error("Exception handling exception= " + tt); } } return true; } ArchivingInputQueueListenerTask(); @Autowired void setCcLocationCache(CcLocationCache cache); @Autowired void setValidationService(RecordValidationService validationService); RealtimeEnvelope deserializeMessage(String contents); @Refreshable(dependsOn = { "inference-engine.inputQueueHost", "inference-engine.inputQueuePort", "inference-engine.inputQueueName"}) void startListenerThread(); @Override String getQueueHost(); @Override String getQueueName(); String getQueueDisplayName(); @Override Integer getQueuePort(); @Override // this method can't throw exceptions or it will stop the queue // listening boolean processMessage(String address, byte[] buff); @PostConstruct void setup(); @PreDestroy void destroy(); @Autowired void setEmergencyStatusNotificationService(
      EmergencyStatusNotificationService emergencyStatusNotificationService); static final int DELAY_THRESHOLD; }### Answer:
@Test public void testProcessMessage() throws IOException { String contents = new BufferedReader(new InputStreamReader(this.getClass().getResourceAsStream("sample-message.json"))).readLine(); t.processMessage("this is the address", contents.getBytes()); }