| /* |
| * Licensed to the Apache Software Foundation (ASF) under one or more |
| * contributor license agreements. See the NOTICE file distributed with |
| * this work for additional information regarding copyright ownership. |
| * The ASF licenses this file to You under the Apache License, Version 2.0 |
| * (the "License"); you may not use this file except in compliance with |
| * the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| */ |
| |
| package org.apache.rocketmq.ons.api.impl.rocketmq; |
| |
| |
| import io.openmessaging.api.Credentials; |
| import io.openmessaging.api.LifeCycle; |
| import java.util.Properties; |
| import java.util.concurrent.ScheduledExecutorService; |
| import java.util.concurrent.ScheduledThreadPoolExecutor; |
| import java.util.concurrent.ThreadFactory; |
| import java.util.concurrent.TimeUnit; |
| import java.util.concurrent.atomic.AtomicBoolean; |
| import javax.annotation.Generated; |
| import org.apache.commons.lang3.StringUtils; |
| import org.apache.rocketmq.client.exception.MQClientException; |
| import org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl; |
| import org.apache.rocketmq.common.UtilAll; |
| import org.apache.rocketmq.common.namesrv.TopAddressing; |
| import org.apache.rocketmq.logging.InternalLogger; |
| import org.apache.rocketmq.ons.api.PropertyKeyConst; |
| import org.apache.rocketmq.ons.api.exception.ONSClientException; |
| import org.apache.rocketmq.ons.api.impl.authority.SessionCredentials; |
| import org.apache.rocketmq.ons.api.impl.util.ClientLoggerUtil; |
| import org.apache.rocketmq.ons.api.impl.util.NameAddrUtils; |
| import org.apache.rocketmq.ons.open.trace.core.dispatch.AsyncDispatcher; |
| |
| import static org.apache.rocketmq.common.UtilAll.getPid; |
| |
| @Generated("ons-client") |
| public abstract class ONSClientAbstract implements LifeCycle, Credentials { |
| |
| protected static final String WSADDR_INTERNAL = System.getProperty("com.aliyun.openservices.ons.addr.internal", |
| "http://onsaddr-internal.aliyun.com:8080/rocketmq/nsaddr4client-internal"); |
| |
| protected static final String WSADDR_INTERNET = System.getProperty("com.aliyun.openservices.ons.addr.internet", |
| "http://onsaddr-internet.aliyun.com/rocketmq/nsaddr4client-internet"); |
| protected static final long WSADDR_INTERNAL_TIMEOUTMILLS = |
| Long.parseLong(System.getProperty("com.aliyun.openservices.ons.addr.internal.timeoutmills", "3000")); |
| protected static final long WSADDR_INTERNET_TIMEOUTMILLS = |
| Long.parseLong(System.getProperty("com.aliyun.openservices.ons.addr.internet.timeoutmills", "5000")); |
| private final static InternalLogger LOGGER = ClientLoggerUtil.getClientLogger(); |
| protected final Properties properties; |
| protected final SessionCredentials sessionCredentials = new SessionCredentials(); |
| protected String nameServerAddr = NameAddrUtils.getNameAdd(); |
| |
| protected AsyncDispatcher traceDispatcher = null; |
| |
| protected final AtomicBoolean started = new AtomicBoolean(false); |
| |
| private final ScheduledExecutorService scheduledExecutorService = new ScheduledThreadPoolExecutor(1, |
| new ThreadFactory() { |
| @Override |
| public Thread newThread(Runnable r) { |
| return new Thread(r, "ONSClient-UpdateNameServerThread"); |
| } |
| }); |
| |
| public ONSClientAbstract(Properties properties) { |
| this.properties = properties; |
| this.sessionCredentials.updateContent(properties); |
| if (this.sessionCredentials.getOnsChannel().equals(ONSChannel.ALIYUN) && |
| (null == this.sessionCredentials.getAccessKey() || "".equals(this.sessionCredentials.getAccessKey()))) { |
| throw new ONSClientException("please set access key"); |
| } |
| |
| if (this.sessionCredentials.getOnsChannel().equals(ONSChannel.ALIYUN) && |
| (null == this.sessionCredentials.getSecretKey() || "".equals(this.sessionCredentials.getSecretKey()))) { |
| throw new ONSClientException("please set secret key"); |
| } |
| |
| if (null == this.sessionCredentials.getOnsChannel()) { |
| throw new ONSClientException("please set ons channel"); |
| } |
| |
| this.nameServerAddr = getNameSrvAddrFromProperties(); |
| if (nameServerAddr != null) { |
| return; |
| } |
| if (nameServerAddr == null && !this.sessionCredentials.getOnsChannel().equals(ONSChannel.ALIYUN)) { |
| return; |
| } |
| this.nameServerAddr = fetchNameServerAddr(); |
| if (null == nameServerAddr) { |
| throw new ONSClientException(FAQ.errorMessage("Can not find name server, May be your network problem.", FAQ.FIND_NS_FAILED)); |
| } |
| |
| this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { |
| @Override |
| public void run() { |
| try { |
| String nsAddrs = fetchNameServerAddr(); |
| if (nsAddrs != null && !ONSClientAbstract.this.nameServerAddr.equals(nsAddrs)) { |
| ONSClientAbstract.this.nameServerAddr = nsAddrs; |
| if (isStarted()) { |
| updateNameServerAddr(nsAddrs); |
| } |
| } |
| } catch (Exception e) { |
| LOGGER.error("update name server periodically failed.", e); |
| } |
| } |
| }, 10 * 1000L, 90 * 1000L, TimeUnit.MILLISECONDS); |
| |
| } |
| |
| protected abstract void updateNameServerAddr(String newAddrs); |
| |
| private String getNameSrvAddrFromProperties() { |
| String nameserverAddrs = this.properties.getProperty(PropertyKeyConst.NAMESRV_ADDR); |
| if (StringUtils.isNotEmpty(nameserverAddrs) && NameAddrUtils.NAMESRV_ENDPOINT_PATTERN.matcher(nameserverAddrs.trim()).matches()) { |
| return nameserverAddrs.substring(NameAddrUtils.ENDPOINT_PREFIX.length()); |
| } |
| |
| return nameserverAddrs; |
| } |
| |
| private String fetchNameServerAddr() { |
| String nsAddrs; |
| |
| { |
| String property = this.properties.getProperty(PropertyKeyConst.ONSAddr); |
| if (property != null) { |
| nsAddrs = new TopAddressing(property).fetchNSAddr(); |
| if (nsAddrs != null) { |
| LOGGER.info("connected to user-defined ons addr server, {} success, {}", property, nsAddrs); |
| return nsAddrs; |
| } else { |
| throw new ONSClientException(FAQ.errorMessage("Can not find name server with onsAddr " + property, FAQ.FIND_NS_FAILED)); |
| } |
| } |
| } |
| |
| { |
| TopAddressing top = new TopAddressing(WSADDR_INTERNAL); |
| nsAddrs = top.fetchNSAddr(false, WSADDR_INTERNAL_TIMEOUTMILLS); |
| if (nsAddrs != null) { |
| LOGGER.info("connected to internal server, {} success, {}", WSADDR_INTERNAL, nsAddrs); |
| return nsAddrs; |
| } |
| } |
| |
| { |
| TopAddressing top = new TopAddressing(WSADDR_INTERNET); |
| nsAddrs = top.fetchNSAddr(false, WSADDR_INTERNET_TIMEOUTMILLS); |
| if (nsAddrs != null) { |
| LOGGER.info("connected to internet server, {} success, {}", WSADDR_INTERNET, nsAddrs); |
| } |
| } |
| |
| return nsAddrs; |
| } |
| |
| public String getNameServerAddr() { |
| return this.nameServerAddr; |
| } |
| |
| protected String buildIntanceName() { |
| StringBuilder stringBuilder = new StringBuilder(); |
| stringBuilder.append(Integer.toString(UtilAll.getPid())).append("#"); |
| if (this.nameServerAddr != null) { |
| stringBuilder.append(this.nameServerAddr.hashCode()).append("#"); |
| } |
| if (this.sessionCredentials.getAccessKey() != null) { |
| stringBuilder.append(this.sessionCredentials.getAccessKey().hashCode()).append("#"); |
| } |
| stringBuilder.append(System.nanoTime()); |
| return stringBuilder.toString(); |
| } |
| |
| protected String getNamespace() { |
| String namespace = null; |
| |
| { |
| String nameserverAddr = this.properties.getProperty(PropertyKeyConst.NAMESRV_ADDR); |
| if (StringUtils.isNotEmpty(nameserverAddr)) { |
| if (NameAddrUtils.validateInstanceEndpoint(nameserverAddr)) { |
| namespace = NameAddrUtils.parseInstanceIdFromEndpoint(nameserverAddr); |
| LOGGER.info("User specify namespace by endpoint: {}.", namespace); |
| } |
| } |
| } |
| |
| { |
| String namespaceFromProperty = this.properties.getProperty(PropertyKeyConst.INSTANCE_ID, null); |
| if (StringUtils.isNotEmpty(namespaceFromProperty)) { |
| namespace = namespaceFromProperty; |
| LOGGER.info("User specify namespace by property: {}.", namespace); |
| } |
| } |
| |
| return namespace; |
| } |
| |
| protected void checkONSProducerServiceState(DefaultMQProducerImpl producer) { |
| switch (producer.getServiceState()) { |
| case CREATE_JUST: |
| throw new ONSClientException( |
| FAQ.errorMessage(String.format("You do not have start the producer[" + getPid() + "], %s", producer.getServiceState()), |
| FAQ.SERVICE_STATE_WRONG)); |
| case SHUTDOWN_ALREADY: |
| throw new ONSClientException(FAQ.errorMessage(String.format("Your producer has been shut down, %s", producer.getServiceState()), |
| FAQ.SERVICE_STATE_WRONG)); |
| case START_FAILED: |
| throw new ONSClientException(FAQ.errorMessage( |
| String.format("When you start your service throws an exception, %s", producer.getServiceState()), FAQ.SERVICE_STATE_WRONG)); |
| case RUNNING: |
| break; |
| default: |
| break; |
| } |
| } |
| |
| @Override |
| public void start() { |
| if (null != traceDispatcher) { |
| try { |
| traceDispatcher.start(); |
| } catch (MQClientException e) { |
| LOGGER.warn("trace dispatcher start failed ", e); |
| } |
| } |
| } |
| |
| @Override |
| public void updateCredential(Properties credentialProperties) { |
| if (this.sessionCredentials.getOnsChannel().equals(ONSChannel.ALIYUN) && |
| (null == credentialProperties.getProperty(SessionCredentials.AccessKey) |
| || "".equals(credentialProperties.getProperty(SessionCredentials.AccessKey)))) { |
| throw new ONSClientException("update credential failed. please set access key."); |
| } |
| |
| if (this.sessionCredentials.getOnsChannel().equals(ONSChannel.ALIYUN) && |
| (null == credentialProperties.getProperty(SessionCredentials.SecretKey) |
| || "".equals(credentialProperties.getProperty(SessionCredentials.SecretKey)))) { |
| throw new ONSClientException("update credential failed. please set secret key"); |
| } |
| this.sessionCredentials.updateContent(credentialProperties); |
| } |
| |
| @Override |
| public void shutdown() { |
| if (null != traceDispatcher) { |
| traceDispatcher.shutdown(); |
| } |
| scheduledExecutorService.shutdown(); |
| } |
| |
| @Override |
| public boolean isStarted() { |
| return started.get(); |
| } |
| |
| @Override |
| public boolean isClosed() { |
| return !isStarted(); |
| } |
| } |
