class HandlerRedirectUtils {

	public static String getRedirectAddress(
			String localJobManagerAddress,
			Tuple2<ActorGateway, Integer> leader) throws Exception {

		final String leaderAddress = leader._1().path();
		final int webMonitorPort = leader._2();

		final String jobManagerName = localJobManagerAddress.substring(localJobManagerAddress.lastIndexOf("/") + 1);

		if (!localJobManagerAddress.equals(leaderAddress) &&
			!leaderAddress.equals(JobManager.getLocalJobManagerAkkaURL(Option.apply(jobManagerName)))) {
			// We are not the leader and need to redirect
			Matcher matcher = LeaderAddressHostPattern.matcher(leaderAddress);

			if (matcher.matches()) {
				String redirectAddress = String.format("%s:%d", matcher.group(1), webMonitorPort);
				return redirectAddress;
			}
			else {
				LOG.warn("Unexpected leader address pattern {}. Cannot extract host.", leaderAddress);
			}
		}

		return null;
	}

}