public void onMessage()

in core/src/main/java/org/apache/cxf/endpoint/ClientImpl.java [821:967]
127 lines of code
28 McCabe index (conditional complexity)

    public void onMessage(Message message) {
        if (bus == null) {
            throw new IllegalStateException("Message received on a Client that has been closed or destroyed.");
        }
        Endpoint endpoint = message.getExchange().getEndpoint();
        if (endpoint == null) {
            // in this case correlation will occur outside the transport,
            // however there's a possibility that the endpoint may have been
            // rebased in the meantime, so that the response will be mediated
            // via a set of in interceptors provided by a *different* endpoint
            //
            endpoint = getConduitSelector().getEndpoint();
            message.getExchange().put(Endpoint.class, endpoint);
        }
        message = endpoint.getBinding().createMessage(message);
        message.getExchange().setInMessage(message);
        message.put(Message.REQUESTOR_ROLE, Boolean.TRUE);
        message.put(Message.INBOUND_MESSAGE, Boolean.TRUE);
        PhaseManager pm = bus.getExtension(PhaseManager.class);

        List<Interceptor<? extends Message>> i1 = bus.getInInterceptors();
        if (LOG.isLoggable(Level.FINE)) {
            LOG.fine("Interceptors contributed by bus: " + i1);
        }
        List<Interceptor<? extends Message>> i2 = getInInterceptors();
        if (LOG.isLoggable(Level.FINE)) {
            LOG.fine("Interceptors contributed by client: " + i2);
        }
        List<Interceptor<? extends Message>> i3 = endpoint.getInInterceptors();
        if (LOG.isLoggable(Level.FINE)) {
            LOG.fine("Interceptors contributed by endpoint: " + i3);
        }
        List<Interceptor<? extends Message>> i4 = endpoint.getBinding().getInInterceptors();
        if (LOG.isLoggable(Level.FINE)) {
            LOG.fine("Interceptors contributed by binding: " + i4);
        }

        PhaseInterceptorChain chain;
        if (endpoint.getService().getDataBinding() instanceof InterceptorProvider) {
            InterceptorProvider p = (InterceptorProvider)endpoint.getService().getDataBinding();
            if (LOG.isLoggable(Level.FINE)) {
                LOG.fine("Interceptors contributed by databinding: " + p.getInInterceptors());
            }
            chain = inboundChainCache.get(pm.getInPhases(), i1, i2, i3, i4,
                                          p.getInInterceptors());
        } else {
            chain = inboundChainCache.get(pm.getInPhases(), i1, i2, i3, i4);
        }
        message.setInterceptorChain(chain);

        chain.setFaultObserver(outFaultObserver);
        modifyChain(chain, message, true);
        modifyChain(chain, message.getExchange().getOutMessage(), true);

        Bus origBus = BusFactory.getAndSetThreadDefaultBus(bus);
        // execute chain
        ClientCallback callback = message.getExchange().get(ClientCallback.class);
        try {
            if (callback != null) {
                if (callback.isCancelled()) {
                    completeExchange(message.getExchange());
                    return;
                }
                callback.start(message);
            }

            String startingAfterInterceptorID = (String) message.get(
                InterceptorChain.STARTING_AFTER_INTERCEPTOR_ID);
            String startingInterceptorID = (String) message.get(
                InterceptorChain.STARTING_AT_INTERCEPTOR_ID);
            if (startingAfterInterceptorID != null) {
                chain.doInterceptStartingAfter(message, startingAfterInterceptorID);
            } else if (startingInterceptorID != null) {
                chain.doInterceptStartingAt(message, startingInterceptorID);
            } else if (message.getContent(Exception.class) != null) {
                outFaultObserver.onMessage(message);
            } else {
                callback = message.getExchange().get(ClientCallback.class);

                if (callback != null && !isPartialResponse(message)) {
                    try {
                        chain.doIntercept(message);
                    } catch (Throwable error) {
                        //so that asyn callback handler get chance to
                        //handle non-runtime exceptions
                        message.getExchange().setInMessage(message);
                        Map<String, Object> resCtx = CastUtils
                                .cast((Map<?, ?>) message.getExchange()
                                        .getOutMessage().get(
                                                Message.INVOCATION_CONTEXT));
                        resCtx = CastUtils.cast((Map<?, ?>) resCtx
                                .get(RESPONSE_CONTEXT));
                        if (resCtx != null) {
                            setResponseContext(resCtx);
                        }
                        // remove callback so that it won't be invoked twice
                        callback = message.getExchange().remove(ClientCallback.class);
                        if (callback != null) {
                            callback.handleException(resCtx, error);
                        }
                    }
                } else {
                    chain.doIntercept(message);
                }

            }

            callback = message.getExchange().get(ClientCallback.class);
            if (callback == null || isPartialResponse(message)) {
                return;
            }

            // remove callback so that it won't be invoked twice
            callback = message.getExchange().remove(ClientCallback.class);
            if (callback != null) {
                message.getExchange().setInMessage(message);
                Map<String, Object> resCtx = CastUtils.cast((Map<?, ?>)message
                                                                .getExchange()
                                                                .getOutMessage()
                                                                .get(Message.INVOCATION_CONTEXT));
                resCtx = CastUtils.cast((Map<?, ?>)resCtx.get(RESPONSE_CONTEXT));
                if (resCtx != null && responseContext != null) {
                    setResponseContext(resCtx);
                }
                try {
                    Object[] obj = processResult(message, message.getExchange(),
                                                 null, resCtx);

                    callback.handleResponse(resCtx, obj);
                } catch (Throwable ex) {
                    callback.handleException(resCtx, ex);
                }
            }
        } finally {
            if (origBus != bus) {
                BusFactory.setThreadDefaultBus(origBus);
            }
            synchronized (message.getExchange()) {
                if (!isPartialResponse(message)
                    || message.getContent(Exception.class) != null) {
                    message.getExchange().put(FINISHED, Boolean.TRUE);
                    message.getExchange().setInMessage(message);
                    message.getExchange().notifyAll();
                }
            }
        }
    }