public static Class<?> extractInputType(Processor<?, ?> processor) {
        Class<?> searchType = processor.getClass();
        while (searchType != Object.class) {
            for (Type t : searchType.getGenericInterfaces()) {
                // if it is a parameterized
                if (t instanceof ParameterizedType && Processor.class.equals(((ParameterizedType) t).getRawType())) {
                    Type inputTypeArgument = ((ParameterizedType) t).getActualTypeArguments()[0];
                    if (inputTypeArgument instanceof ParameterizedType) {
                        return (Class) ((ParameterizedType) inputTypeArgument)
                                .getRawType();
                    } else if (inputTypeArgument instanceof Class) {
                        return (Class) inputTypeArgument;
                    }
                } else if (t instanceof Class && Processor.class.equals(t)) {
                    return Object.class;
                }
            }
            searchType = searchType.getSuperclass();
        }
        throw new IllegalStateException("Cannot identify the input type");
    }
Given the processor interface, determine it's input type parameter

@param processor and instance of the processor interface
@return the corresponding Class of the processor input type parameter, or null if can't
be determined.
spring-xd-reactor/src/main/java/org/springframework/xd/reactor/ReactorReflectionUtils.java