Heating extends BaseOpenMessage { public static Func1<OpenSession, OpenSession> handleTemperature(Action1<Double> onSuccess, Action0 onError) { return openSession -> { isValidPrefixType(openSession.getRequest(), FORMAT_PREFIX_DIMENSION_WHO, WHO); List<OpenMessage> response = openSession.getResponse(); checkNotNull(response, "response is null"); checkArgument(response.size() == 1 || response.size() == 2, "invalid response"); final String responseValue = response.get(0).getValue(); checkNotNull(responseValue, "response value is null"); if (isValidTemperature(responseValue)) { onSuccess.call(getTemperature(openSession.getRequest(), responseValue)); } else { onError.call(); } return openSession; }; } private  Heating(String value, TemperatureScale scale); static Heating requestTemperature(String where, TemperatureScale scale); static Heating requestTemperature(String where); static Func1<OpenSession, OpenSession> handleTemperature(Action1<Double> onSuccess, Action0 onError);  }
@Test public void testHandleTemperature() { Action1 onSuccessMock = mock(Action1.class); Action0 onErrorMock = mock(Action0.class); OpenSession openSession = OpenSession.newSession(requestTemperature("0")); openSession.addAllResponse(Lists.newArrayList(() -> "*#4*0*0*0225##", () -> OpenMessage.ACK)); Observable.just(openSession) .map(Heating.handleTemperature(onSuccessMock, onErrorMock)) .subscribe(); verify(onSuccessMock).call(new Double(22.5)); verify(onErrorMock, never()).call(); }