import org.apache.camel.Headers; //导入依赖的package包/类
/**
 * Create a ServerAuthorization object.
 * 
 * @param obj
 * @param respHdrs
 * @return
 */
public final ServerAuthorization create(ServerAuthorization serverAuth,
    @Headers Map<String, Object> reqHdrs,
    @OutHeaders Map<String, Object> respHdrs) {

  final String serverUrl = serverAuth.getServerUrl();
  if (serverUrl != null && !serverUrl.isEmpty()) {
    // Don't honor the incoming id value, if any
    serverAuth.setId(UUID.randomUUID().toString());

    final ServerAuthorization serverAuthResp = processServerAuthRequest(
        serverAuth, reqHdrs, respHdrs);

    return serverAuthResp;
  }
  return null;
}
 

import org.apache.camel.Headers; //导入依赖的package包/类
@Consume(uri = "activemq:Test.BindingQueue")
public void myMethod(@Headers Map<?, ?> headers, String body) {
    this.headers = headers;
    this.body = body;

    // now lets notify we've completed
    producer.sendBody("Completed");
}
 

import org.apache.camel.Headers; //导入依赖的package包/类
/**
 * Partially reads the input message and determines what type of message this is
 *
 * (right now this it pretty dumb the goal for this would to replace this with some sort of message codec)
 *
 * @param input  - the message input stream
 * @param headers- the message headers that we will add the additional headers to contain the message version details
 * @throws IOException
 */
public void parse(@Body InputStream input, @Headers Map<String, Object> headers) throws IOException {
    try {
        byte [] msgData = new byte[20480];
        int count = input.read(msgData);
        if(count > 0) {
            String msg = new String(msgData); //should be able to use a header to determine encoding

            //First determine if the message is a SOAP 1.1 or 1.2 message by default we will assume 1.1
            String soapVersion = msg.contains(EbmsConstants.SOAP_1_2_NAMESPACE) ? SOAPConstants.SOAP_1_2_PROTOCOL : SOAPConstants.SOAP_1_1_PROTOCOL;
            headers.put(EbmsConstants.SOAP_VERSION,soapVersion);

            //next determine what version of ebms message is it, by default assume ebms V2
            String ebmsVersion = msg.contains(EbmsConstants.EBXML_V3_NAMESPACE) ? EbmsConstants.EBMS_V3 : EbmsConstants.EBMS_V2;
            headers.put(EbmsConstants.EBMS_VERSION,ebmsVersion);

            headers.put(EbmsConstants.MESSAGE_ID, getMessageId(msg));
            headers.put(EbmsConstants.REF_TO_MESSAGE_ID, getRefMessageId(msg));
            headers.put(EbmsConstants.MESSAGE_TYPE, getMessageType(msg).name());
        }
    } finally {
        input.reset();
    }
}
 

import org.apache.camel.Headers; //导入依赖的package包/类
@SuppressWarnings("unused")
public String forward(String body,
					  @Headers Map<String, Object> headers,
					  @Properties Map<String, Object> properties,
					  @Header(Exchange.SLIP_ENDPOINT) String previous) {
	
	if (previous == null) {
		Object routing = headers.get("router-header");
		return routing != null ? ("direct:" + routing) : null;
	}
	
	// no more so return null
	return null;
}
 

import org.apache.camel.Headers; //导入依赖的package包/类
/**
 * Process a request to create a Server Authorization (i.e., request to grant
 * ptmatchadapter authorization to access a particular fhir server)
 * 
 * @param serverAuth
 * @param reqHdrs
 * @param respHdrs
 * @return
 */
private final ServerAuthorization processServerAuthRequest(
    ServerAuthorization serverAuth,
    @Headers Map<String, Object> reqHdrs,
    @OutHeaders Map<String, Object> respHdrs) {
  final String serverUrl = serverAuth.getServerUrl();
  final String accessToken = serverAuth.getAccessToken();

  // if request doesn't contain a server URL, it is an error
  if (serverUrl == null || serverUrl.isEmpty()) {
    respHdrs.put(Exchange.HTTP_RESPONSE_CODE, 400); // BAD REQUEST
    return null;
  }
  // else if the request body doesn't include an access token, redirect user
  // to an authorization server
  else if (accessToken == null || accessToken.isEmpty()) {
    // create a state identifier
    final String stateKey = newStateKey();

    respHdrs.put(STATE_PARAM, stateKey);

    final AuthorizationRequestInfo requestInfo = new AuthorizationRequestInfo();
    requestInfo.put(SERVER_AUTH, serverAuth);
    sessionData.put(stateKey, requestInfo);

    // Construct URL we will invoke on authorization server
    // GET /authorize?response_type=code&client_id=s6BhdRkqt3&state=xyz
    // &redirect_uri=https%3A%2F%2Fclient%2Eexample%2Ecom%2Fcb
    final StringBuilder authUrl = new StringBuilder(100);
    if (getAuthorizationServer() != null) {
      authUrl.append(getAuthorizationServer());
    }
    authUrl.append(getAuthorizationEndpoint());
    authUrl.append("?");

    authUrl.append("response_type=code&client_id=");
    try {
      authUrl.append(URLEncoder.encode(getClientId(), "UTF-8"));
      authUrl.append("&");
      authUrl.append(STATE_PARAM);
      authUrl.append("=");
      authUrl.append(stateKey);
      authUrl.append("&redirect_uri=");

      final HttpServletRequest req = (HttpServletRequest) reqHdrs
          .get(Exchange.HTTP_SERVLET_REQUEST);
      final String redirectUri = URLEncoder.encode(
          getClientAuthRedirectUri(req.getScheme(), req.getServerName(),
              req.getServerPort()),
          "UTF-8");
      authUrl.append(redirectUri);
      // we need to provide redirect uri with access token request, so save it
      requestInfo.put("redirectUri", redirectUri);
    } catch (UnsupportedEncodingException e) {
      // Should never happen, which is why I wrap all above once
      LOG.error("Usupported encoding used on authorization redirect", e);
    }

    respHdrs.put(Exchange.HTTP_RESPONSE_CODE, 302); // FOUND
    respHdrs.put(Exchange.CONTENT_TYPE, "text/plain");
    respHdrs.put("Location", authUrl.toString());

    return null;
  } else {
    LOG.warn("NOT IMPLEMENTED");
    return null;
  }
}
 

import org.apache.camel.Headers; //导入依赖的package包/类
public void prepareJdbcHeaders(@Body Measure measure, @Headers Map<String, Object> headers) {

    	headers.put("sensor_type", measure.getSensorType());
    	headers.put("data_type", measure.getDataType());
    	headers.put("device_id", measure.getDeviceId());
    	headers.put("category", measure.getCategory());
    	headers.put("payload", measure.getPayload());
    	headers.put("error_code", measure.getErrorCode());
    	headers.put("error_message", measure.getErrorMessage());
    	headers.put("time_stamp", measure.getTimestamp());

    }
 

import org.apache.camel.Headers; //导入依赖的package包/类
/**
 * Use this method to compute dynamic where we should route next.
 *
 * @param body the message body
 * @param headers the message headers where we can store state between invocations
 * @param previous the previous slip
 * @return endpoints to go, or <tt>null</tt> to indicate the end
 */
public String slip(String body, @Headers Map<String, Object> headers, @Header(Exchange.SLIP_ENDPOINT) String previous) {
    bodies.add(body);
    if (previous != null) {
        previouses.add(previous);
    }

    // get the state from the message headers and keep track how many times
    // we have been invoked
    int invoked = 0;
    Object current = headers.get("invoked");
    if (current != null) {
        invoked = Integer.valueOf(current.toString());
    }
    invoked++;
    // and store the state back on the headers
    headers.put("invoked", invoked);

    if (invoked == 1) {
        return "mock:a";
    } else if (invoked == 2) {
        return "mock:b,mock:c";
    } else if (invoked == 3) {
        return "direct:foo";
    } else if (invoked == 4) {
        return "mock:result";
    }

    // no more so return null
    return null;
}
 

import org.apache.camel.Headers; //导入依赖的package包/类
/**
 * This method handle our order input and return the order
 *
 * @param in      the in headers
 * @param payload the in payload
 * @param out     the out headers
 * @return the out payload
 * @throws OrderFailedException is thrown if the order cannot be processed
 */
public Object handleOrder(@Headers Map<?, ?> in, @Body String payload, @OutHeaders Map<String, Object> out)
    throws OrderFailedException {
    out.put("customerid", in.get("customerid"));
    if ("Order: kaboom".equals(payload)) {
        throw new OrderFailedException("Cannot order: kaboom");
    } else {
        out.put("orderid", "123");
        return "Order OK";
    }
}
 

import org.apache.camel.Headers; //导入依赖的package包/类
/**
 * Use this method to compute dynamic where we should route next.
 *
 * @param body the message body
 * @param headers the message headers where we can store state between invocations
 * @return endpoints to go, or <tt>null</tt> to indicate the end
 */
public String slip(String body, @Headers Map<String, Object> headers) {
    bodies.add(body);

    // get the state from the message headers and keep track how many times
    // we have been invoked
    int invoked = 0;
    Object current = headers.get("invoked");
    if (current != null) {
        invoked = Integer.valueOf(current.toString());
    }
    invoked++;
    // and store the state back on the headers
    headers.put("invoked", invoked);

    if (invoked == 1) {
        return "mock:a";
    } else if (invoked == 2) {
        return "mock:b,mock:c";
    } else if (invoked == 3) {
        return "direct:foo";
    } else if (invoked == 4) {
        return "mock:result";
    }

    // no more so return null
    return null;
}
 

import org.apache.camel.Headers; //导入依赖的package包/类
public void myMethod(@Properties Map<?, ?> foo, @Headers Map<?, ?> bar, @Body String body) {
    this.foo = foo;
    this.bar = bar;
    this.body = body;
    
    assertNotNull(toString());
}
 

import org.apache.camel.Headers; //导入依赖的package包/类
public String doSomething(@Body String body, @Headers Map<?, ?> headers,
                          @OutHeaders Map<String, Object> outHeaders) {
    if (outHeaders != null) {
        outHeaders.put("out", 123);
    }

    return "Hello!";
}
 

import org.apache.camel.Headers; //导入依赖的package包/类
/**
 * This method handle our order input and return the order
 * 
 * @param in the in headers
 * @param payload the in payload
 * @param out the out headers
 * @return the out payload
 * @throws OrderFailedException is thrown if the order cannot be processed
 */
public Object handleOrder(@Headers Map<String, Object> in, @Body String payload, @OutHeaders Map<String, Object> out)
    throws OrderFailedException {
    out.put("customerid", in.get("customerid"));
    if ("Order: kaboom".equals(payload)) {
        throw new OrderFailedException("Cannot order: kaboom");
    } else {
        out.put("orderid", "123");
        return "Order OK";
    }
}
 

import org.apache.camel.Headers; //导入依赖的package包/类
public String hello(@Body String body, @Header("foo") String foo, @Headers Map<String, Object> headers) {
    String s = body.replaceFirst("Hello", "Bye");

    if (!foo.equals("ABC")) {
        throw new IllegalArgumentException("Foo has not expected value ABC but " + foo);
    }

    headers.put("foo", 123);
    return s;
}
 

import org.apache.camel.Headers; //导入依赖的package包/类
public String toSql(@XPath("order/@name") String name,
                    @XPath("order/@amount") int amount,
                    @XPath("order/@customer") String customer,
                    @Headers Map<String, Object> outHeaders) {
    outHeaders.put("partName", name);
    outHeaders.put("quantity", amount);
    outHeaders.put("customer", customer);
    return "insert into incoming_orders (part_name, quantity, customer) values (:?partName, :?quantity, :?customer)";
}
 

import org.apache.camel.Headers; //导入依赖的package包/类
/**
 * Sets {@link AsynchConstants#OBJECT_ID_HEADER} and {@link AsynchConstants#ENTITY_TYPE_HEADER}
 * headers if there are available corresponding values in message.
 *
 * @param msg the message
 * @param headers the headers
 */
@Handler
public void setEntityInfo(@Body Message msg, @Headers Map<String, Object> headers) {
    Assert.notNull(msg, "the msg must not be null");

    if (msg.getObjectId() != null) {
        headers.put(OBJECT_ID_HEADER, msg.getObjectId());
    }
    if (msg.getEntityType() != null) {
        headers.put(ENTITY_TYPE_HEADER, msg.getEntityType());
    }
}
 

import org.apache.camel.Headers; //导入依赖的package包/类
@Consume(uri = "direct:start")
@RoutingSlip(delimiter = ",")
public List<String> routeMe(String body, @Headers Map<String, Object> headers) {
    ArrayList<String> results = new ArrayList<String>();

    Object slip = headers.get("myRoutingSlipHeader");
    if (slip != null) {
        String[] uris = slip.toString().split(",");
        Collections.addAll(results, uris);
    }

    results.add("mock:oneMore");

    return results;
}
 

import org.apache.camel.Headers; //导入依赖的package包/类
public void logHeadersByPattern(@Headers Map<String,Object> headers){
    if (logHeadersPattern != null) {
        log.info("Headers: {}", headers.keySet().stream().filter(s -> s.matches(logHeadersPattern)).collect(Collectors.toList()));
    }
}
 

import org.apache.camel.Headers; //导入依赖的package包/类
/**
 * Processes a form-based request to create a ServerAuthorization
 * 
 * @param body
 *          Body of the request (unused since form parameters are expected in
 *          the request header
 * @param reqHdrs
 * @param respHdrs
 * @return
 */
public final ServerAuthorization createFromForm(@Body String body,
    @Headers Map<String, Object> reqHdrs,
    @OutHeaders Map<String, Object> respHdrs) {


  final String serverUrl = (String) reqHdrs.get("serverUrl");
  if (serverUrl != null && !serverUrl.isEmpty()) {
    final ServerAuthorization serverAuth = new ServerAuthorization();
    serverAuth.setId(UUID.randomUUID().toString());
    serverAuth.setTitle((String) reqHdrs.get("title"));
    serverAuth.setServerUrl(serverUrl);

    // look for evidence of CORS header (header is case-insensitive
    String origin = (String) reqHdrs.get("Origin");
    if (origin == null) {
      origin = (String) reqHdrs.get("origin");
    }
    LOG.debug("handleOptions: origin {}", origin);

    // Section 3.2 of RFC 7230 (https://tools.ietf.org/html/rfc7230#section-3.2)
    // says header fields are case-insensitive
    if (origin != null) {
      // Firefox on Linux wan'ts exact value of origin in response; * is being rejected
      respHdrs.put("Access-Control-Allow-Origin", origin);
      respHdrs.put("Access-Control-Allow-Credentials", "true");
    }
    
    // Redirect caller to authorization server to get an authorization code
    final ServerAuthorization serverAuthResp = processServerAuthRequest(
        serverAuth, reqHdrs, respHdrs);

    // Retrieve the state key from the query parameters
    final String stateKey = (String) respHdrs.get(STATE_PARAM);

    final AuthorizationRequestInfo requestInfo = (AuthorizationRequestInfo) sessionData
        .get(stateKey);

    // Annotate request info so we know to return html later
    requestInfo.setResponseType("html");

    return serverAuthResp;
  } else {
    // missing required parameter
    respHdrs.put(Exchange.HTTP_RESPONSE_CODE, 400); // BAD REQUEST
    respHdrs.put(Exchange.CONTENT_LENGTH, 0);
  }
  return null;
}
 

import org.apache.camel.Headers; //导入依赖的package包/类
/**
 * Processes authorization code response from the OAuth 2.0 Authorization
 * Server.
 * 
 * @param body
 * @param reqHdrs
 * @param respHdrs
 * @return
 */
public String processAuthorizationCode(@Body String body,
    @Headers Map<String, Object> reqHdrs,
    @OutHeaders Map<String, Object> respHdrs) {
  // Retrieve the state key from the query parameters
  final String stateKey = (String) reqHdrs.get(STATE_PARAM);

  if (stateKey == null) {
    final String msg = "Redirect from authorization server is missing state parameter";
    LOG.error(msg);
    throw new IllegalStateException(msg);
  }

  LOG.info("process redirect, state {}", stateKey);

  for (String key : sessionData.keySet()) {
    LOG.info("redirect session state key: {}", key);
  }

  final HttpServletRequest req = (HttpServletRequest) reqHdrs
      .get(Exchange.HTTP_SERVLET_REQUEST);
  
  final String authCode = (String) reqHdrs.get(CODE_PARAM);

  // - - - - - - - - - - - - - - - - - - - - - - - - - -
  // Request an Access Token from the OAuth Authorization Server
  // - - - - - - - - - - - - - - - - - - - - - - - - - -
  final ServerAuthorization serverAuth = requestAccessToken(req, stateKey,
      authCode);

  if (serverAuth != null) {
    getServerAuthorizations().add(serverAuth);
    LOG.info("process AuthCodeResp, serverUrl {}", serverAuth.getServerUrl());
    LOG.info("process AuthCodeResp, # server auths {}",
        serverAuthorizations.size());
  }

  final AuthorizationRequestInfo requestInfo = (AuthorizationRequestInfo) sessionData
      .remove(stateKey);

  if (requestInfo.getResponseType().equalsIgnoreCase("html")) {
    // redirect user to page of server authorizations
    respHdrs.put(Exchange.HTTP_RESPONSE_CODE, 302); // FOUND
    respHdrs.put(Exchange.CONTENT_LENGTH, 0);
    respHdrs.put("Location", "/");
    return "";
  } else {
    respHdrs.put(Exchange.HTTP_RESPONSE_CODE, 201); // Created
    respHdrs.put(Exchange.CONTENT_TYPE, "application/json");
    return "{\"code\": \"success\"}";
  }
}
 

import org.apache.camel.Headers; //导入依赖的package包/类
public String doSomething(String body, @Headers Map<Object, Object> headers) throws Exception {
    headers.put("foo", foo);
    return "Hello " + body;
}