src/rdkafka_broker.c
static int rd_kafka_broker_name_parse (rd_kafka_t *rk,
				       char **name,
				       rd_kafka_secproto_t *proto,
				       const char **host,
				       uint16_t *port) {
	char *s = *name;
	char *orig;
	char *n, *t, *t2;

	/* Save a temporary copy of the original name for logging purposes */
	rd_strdupa(&orig, *name);

	/* Find end of this name (either by delimiter or end of string */
	if ((n = strchr(s, ',')))
		*n = '\0';
	else
		n = s + strlen(s)-1;


	/* Check if this looks like an url. */
	if ((t = strstr(s, "://"))) {
		int i;
		/* "proto://host[:port]" */

		if (t == s) {
			rd_kafka_log(rk, LOG_WARNING, "BROKER",
				     "Broker name \"%s\" parse error: "
				     "empty protocol name", orig);
			return -1;
		}

		/* Make protocol uppercase */
		for (t2 = s ; t2 < t ; t2++)
			*t2 = toupper(*t2);

		*t = '\0';

		/* Find matching protocol by name. */
		for (i = 0 ; i < RD_KAFKA_PROTO_NUM ; i++)
			if (!rd_strcasecmp(s, rd_kafka_secproto_names[i]))
				break;

		/* Unsupported protocol */
		if (i == RD_KAFKA_PROTO_NUM) {
			rd_kafka_log(rk, LOG_WARNING, "BROKER",
				     "Broker name \"%s\" parse error: "
				     "unsupported protocol \"%s\"", orig, s);

			return -1;
		}

		*proto = i;

                /* Enforce protocol */
		if (rk->rk_conf.security_protocol != *proto) {
			rd_kafka_log(rk, LOG_WARNING, "BROKER",
				     "Broker name \"%s\" parse error: "
				     "protocol \"%s\" does not match "
				     "security.protocol setting \"%s\"",
				     orig, s,
				     rd_kafka_secproto_names[
					     rk->rk_conf.security_protocol]);
			return -1;
		}

		/* Hostname starts here */
		s = t+3;

		/* Ignore anything that looks like the path part of an URL */
		if ((t = strchr(s, '/')))
			*t = '\0';

	} else
		*proto = rk->rk_conf.security_protocol; /* Default protocol */


	*port = RD_KAFKA_PORT;
	/* Check if port has been specified, but try to identify IPv6
	 * addresses first:
	 *  t = last ':' in string
	 *  t2 = first ':' in string
	 *  If t and t2 are equal then only one ":" exists in name
	 *  and thus an IPv4 address with port specified.
	 *  Else if not equal and t is prefixed with "]" then it's an
	 *  IPv6 address with port specified.
	 *  Else no port specified. */
	if ((t = strrchr(s, ':')) &&
	    ((t2 = strchr(s, ':')) == t || *(t-1) == ']')) {
		*t = '\0';
		*port = atoi(t+1);
	}

	/* Empty host name -> localhost */
	if (!*s) 
		s = "localhost";

	*host = s;
	*name = n+1;  /* past this name. e.g., next name/delimiter to parse */

	return 0;
}
/**
 * Parse a broker host name.
 * The string 'name' is modified and null-terminated portions of it
 * are returned in 'proto', 'host', and 'port'.
 *
 * Returns 0 on success or -1 on parse error.
 */
Parse a broker host name.

Returns 0 on success or -1 on parse error.
Parse a broker host name . Returns 0 on success or - 1 on parse error .
static int rd_kafka_broker_name_parse (rd_kafka_t *rk,
				       char **name,
				       rd_kafka_secproto_t *proto,
				       const char **host,
				       uint16_t *port) {
	char *s = *name;
	char *orig;
	char *n, *t, *t2;
	rd_strdupa(&orig, *name);
	if ((n = strchr(s, ',')))
		*n = '\0';
	else
		n = s + strlen(s)-1;
	if ((t = strstr(s, "://"))) {
		int i;
		if (t == s) {
			rd_kafka_log(rk, LOG_WARNING, "BROKER",
				     "Broker name \"%s\" parse error: "
				     "empty protocol name", orig);
			return -1;
		}
		for (t2 = s ; t2 < t ; t2++)
			*t2 = toupper(*t2);
		*t = '\0';
		for (i = 0 ; i < RD_KAFKA_PROTO_NUM ; i++)
			if (!rd_strcasecmp(s, rd_kafka_secproto_names[i]))
				break;
		if (i == RD_KAFKA_PROTO_NUM) {
			rd_kafka_log(rk, LOG_WARNING, "BROKER",
				     "Broker name \"%s\" parse error: "
				     "unsupported protocol \"%s\"", orig, s);
			return -1;
		}
		*proto = i;
		if (rk->rk_conf.security_protocol != *proto) {
			rd_kafka_log(rk, LOG_WARNING, "BROKER",
				     "Broker name \"%s\" parse error: "
				     "protocol \"%s\" does not match "
				     "security.protocol setting \"%s\"",
				     orig, s,
				     rd_kafka_secproto_names[
					     rk->rk_conf.security_protocol]);
			return -1;
		}
		s = t+3;
		if ((t = strchr(s, '/')))
			*t = '\0';
	} else
		*proto = rk->rk_conf.security_protocol; 
	*port = RD_KAFKA_PORT;
	if ((t = strrchr(s, ':')) &&
	    ((t2 = strchr(s, ':')) == t || *(t-1) == ']')) {
		*t = '\0';
		*port = atoi(t+1);
	}
	if (!*s) 
		s = "localhost";
	*host = s;
	*name = n+1;  
	return 0;
}
static int rd_kafka_broker_name_parse ( rd_kafka_t * rk , char * * name , rd_kafka_secproto_t * proto , const char * * host , uint16_t * port ) { char * s = * name ; char * orig ; char * n , * t , * t2 ; rd_strdupa ( & orig , * name ) ; if ( ( n = strchr ( s , ' ' ) ) ) * n = ' \0 ' ; else n = s + strlen ( s ) - 1 ; if ( ( t = strstr ( s , " " ) ) ) { int i ; if ( t == s ) { rd_kafka_log ( rk , LOG_WARNING , " " , " \" \" " " " , orig ) ; return -1 ; } for ( t2 = s ; t2 < t ; t2 ++ ) * t2 = toupper ( * t2 ) ; * t = ' \0 ' ; for ( i = 0 ; i < RD_KAFKA_PROTO_NUM ; i ++ ) if ( ! rd_strcasecmp ( s , rd_kafka_secproto_names [ i ] ) ) break ; if ( i == RD_KAFKA_PROTO_NUM ) { rd_kafka_log ( rk , LOG_WARNING , " " , " \" \" " " \" \" " , orig , s ) ; return -1 ; } * proto = i ; if ( rk -> rk_conf . security_protocol != * proto ) { rd_kafka_log ( rk , LOG_WARNING , " " , " \" \" " " \" \" " " \" \" " , orig , s , rd_kafka_secproto_names [ rk -> rk_conf . security_protocol ] ) ; return -1 ; } s = t + 3 ; if ( ( t = strchr ( s , ' ' ) ) ) * t = ' \0 ' ; } else * proto = rk -> rk_conf . security_protocol ; * port = RD_KAFKA_PORT ; if ( ( t = strrchr ( s , ' ' ) ) && ( ( t2 = strchr ( s , ' ' ) ) == t || * ( t - 1 ) == ' ' ) ) { * t = ' \0 ' ; * port = atoi ( t + 1 ) ; } if ( ! * s ) s = " " ; * host = s ; * name = n + 1 ; return 0 ; }
Parse a broker host name.
Parse a broker host name .
/* Save a temporary copy of the original name for logging purposes */ /* Find end of this name (either by delimiter or end of string */ /* Check if this looks like an url. */ /* "proto://host[:port]" */ /* Make protocol uppercase */ /* Find matching protocol by name. */ /* Unsupported protocol */ /* Enforce protocol */ /* Hostname starts here */ /* Ignore anything that looks like the path part of an URL */ /* Default protocol */ /* Check if port has been specified, but try to identify IPv6
	 * addresses first:
	 *  t = last ':' in string
	 *  t2 = first ':' in string
	 *  If t and t2 are equal then only one ":" exists in name
	 *  and thus an IPv4 address with port specified.
	 *  Else if not equal and t is prefixed with "]" then it's an
	 *  IPv6 address with port specified.
	 *  Else no port specified. */ /* Empty host name -> localhost */ /* past this name. e.g., next name/delimiter to parse */