选择视图同步

@param channelInfo
@param channelParameterInfo
@throws Exception
187829
public void doNextToView(@FormGroup("dataMediaPairInfo") Group dataMediaPairInfo,
                             @FormField(name = "formDataMediaPairError", group = "dataMediaPairInfo") CustomErrors err,
                             Navigator nav) throws Exception {
        DataMediaPair dataMediaPair = new DataMediaPair();
        DataMedia sourceDataMedia = new DataMedia();
        DataMedia targetDataMedia = new DataMedia();
        dataMediaPairInfo.setProperties(dataMediaPair);
        sourceDataMedia.setId(dataMediaPairInfo.getField("sourceDataMediaId").getLongValue());
        dataMediaPair.setSource(sourceDataMedia);
        targetDataMedia.setId(dataMediaPairInfo.getField("targetDataMediaId").getLongValue());
        dataMediaPair.setTarget(targetDataMedia);
        try {
            dataMediaPairService.create(dataMediaPair);
        } catch (RepeatConfigureException rce) {
            err.setMessage("invalidDataMediaPair");
            return;
        }

        nav.redirectToLocation("dataMediaPairList.htm?pipelineId=" + dataMediaPair.getPipelineId());
    }
1031890
@Process(actionType = CreateHandler.class)
    public void createHandler(final CreateHandler action, final Dispatcher.Channel channel) {
        operationDelegate.onCreateResource(action.getAddressTemplate(), action.getInstanceName(), action.getNewModel(),
            new CrudOperationDelegate.Callback() {
                @Override
                public void onFailure(AddressTemplate template, String name, Throwable t) {
                    channel.nack(t);
                }

                @Override
                public void onSuccess(AddressTemplate addressTemplate, String name) {
                    lastModifiedInstance = action.getInstanceName();
                    // we must strip the pre-handler-chain suffix to reload a single configuration. 
                    readConfig(action.getAddressTemplate().subTemplate(0, 3), channel);
                }
            });
    }
398225
public void doDelete(@Param("channelId") Long channelId, @Param("pageIndex") int pageIndex,
                         @Param("searchKey") String searchKey, Navigator nav) throws WebxException {
        if (channelService.findById(channelId).getStatus().isStart()) {
            nav.redirectTo(WebConstant.ERROR_FORBIDDEN_Link);
            return;
        }

        // 如果channel节点下面还有关联的pipeline时，不允许删除
        if (pipelineService.listByChannelIds(channelId).size() < 1) {
            channelService.remove(channelId);
        }

        nav.redirectToLocation("channelList.htm?pageIndex=" + pageIndex + "&searchKey=" + urlEncode(searchKey));
    }
166059
public void execute(@Param("channelId") Long channelId, @Param("pageIndex") int pageIndex,
                        @Param("searchKey") String searchKey, Context context, Navigator nav) throws Exception {
        Channel channel = channelService.findById(channelId);
        if (channel.getStatus().isStart()) {
            nav.redirectTo(WebConstant.ERROR_FORBIDDEN_Link);
            return;
        }
        context.put("channel", channel);
        context.put("pageIndex", pageIndex);
        context.put("searchKey", searchKey);
    }
440329
protected ChannelHandler initLogHandlerAdapter(ChannelHandler init)
	{
		ChannelHandler handler=new  ShareableChannelInboundHandler() {
			@Override
			public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
				Channel ch=ctx.channel();
				manageChannel(ch);
				LogLevel level=config.getChannelLevel();
				if(level!=null)
				{//单个链路的日志记录器
					ch.pipeline().addLast(new LoggerHandler(level));
				}
				ch.pipeline().addLast(init);
				ctx.pipeline().remove(this);//移除当前handler
				ctx.fireChannelRegistered();//从当前handler往后抛出事件
			}
		};
//		ChannelHandler handler=new ChannelInitializer<Channel>() {
//			@Override
//			protected void initChannel(Channel ch) throws Exception {
//				channelGroup.add(ch);
//				LogLevel level=config.getLevel();
//				if(level!=null)
//				{
//					ch.pipeline().addLast(new LoggerHandler(config.getLevel()));
//				}
//				ch.pipeline().addLast(init);
//			}
//		};
		return handler;
	}
475654
private Map getPublishedChannelParametersMap(String channelPublishId) throws PortalException {
        try {
            IPortletDefinitionRegistry registry =
                    PortletDefinitionRegistryLocator.getPortletDefinitionRegistry();
            IPortletDefinition def = registry.getPortletDefinition(channelPublishId);
            return def.getParametersAsUnmodifiableMap();
        } catch (Exception e) {
            throw new PortalException("Unable to acquire channel definition.", e);
        }
    }
1314484
public void update(ChannelData channelData) {
        synchronized (this.udpChannelConfig) {
            // can't do two updates at the same time
            try {
                this.udpChannelConfig.setChannelData(channelData);
            } catch (ChannelException e) {
                if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                    Tr.debug(this, tc, "Error updating config:" + e.getMessage());
                }
            }
        }
    }
406399
@Skip
    @Override
    public void bind(ChannelHandlerContext ctx, SocketAddress localAddress,
                     ChannelPromise promise) throws Exception {
        ctx.bind(localAddress, promise);
    }
962679
protected void handleHttpRequest(final ChannelHandlerContext channelContext, final HttpRequest request) throws Exception {
        if (isWebSocketsUpgradeRequest( request )) {

            final Handshake handshake = findHandshake( request );

            if (handshake != null) {

                HttpResponse response = handshake.generateResponse( request );

                response.addHeader( Names.UPGRADE, Values.WEBSOCKET );
                response.addHeader( Names.CONNECTION, Values.UPGRADE );

                final ChannelPipeline pipeline = channelContext.getChannel().getPipeline();
                reconfigureUpstream( pipeline, handshake );

                Channel channel = channelContext.getChannel();
                ChannelFuture future = channel.write( response );
                future.addListener( new ChannelFutureListener() {
                    public void operationComplete(ChannelFuture future) throws Exception {
                        reconfigureDownstream( pipeline, handshake );
                        pipeline.replace( ServerHandshakeHandler.this, "websocket-disconnection-negotiator", new WebSocketDisconnectionNegotiator() );
                        forwardConnectEventUpstream( channelContext );
                        decodeHost( channelContext, request );
                        decodeSession( channelContext, request );
                    }
                } );

                return;
            }
        }

        // Send an error page otherwise.
        sendHttpResponse( channelContext, request, new DefaultHttpResponse( HttpVersion.HTTP_1_1, HttpResponseStatus.FORBIDDEN ) );
    }
744438
@Override
  public void exceptionCaught(final ChannelHandlerContext ctx, final Throwable cause) {
    final Channel channel = ctx.channel();

    LOG.log(Level.INFO, "Unexpected exception from downstream. channel: {0} local: {1} remote: {2}",
        new Object[]{channel, channel.localAddress(), channel.remoteAddress()});
    LOG.log(Level.WARNING, "Unexpected exception from downstream.", cause);
    channel.close();
    this.listener.exceptionCaught(ctx, cause);
  }
1362851
public void init() throws ChannelException {
        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
            Tr.debug(this, tc, "UDP Channel: " + getConfig().getExternalName() + " listening port: " + String.valueOf(getConfig().getPort()));
        }
    }
1023882
public void init() throws ChannelException {
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) {
            Tr.entry(tc, "init");
        }
        super.init();

        if (!getConfig().isInbound()) {
            try {
                this.asyncChannelGroup = findOrCreateACG();
                if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                    Tr.debug(tc, "Created completion port for outbound connections, completionPort = " + this.asyncChannelGroup.getCompletionPort());
                }
            } catch (AsyncException ae) {
                ChannelException ce = new ChannelException("Error creating async channel group ");
                ce.initCause(ae);
                throw ce;
            }
        }
        // create AIO CompletionListeners
        aioReadCompletionListener = new AioReadCompletionListener();
        aioWriteCompletionListener = new AioWriteCompletionListener();

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) {
            Tr.exit(tc, "init");
        }
    }
436475
synchronized void recoverChannel(boolean viaConnectionRecovery) throws Exception {
    recoveryPending.set(false);
    if (circuit.isClosed())
      return;

    if (recoveryStats == null) {
      recoveryConsumers = consumerDeclarations.isEmpty() ? null
        : new LinkedHashMap<String, ConsumerDeclaration>(consumerDeclarations);
      recoveryStats = new RecurringStats(config.getChannelRecoveryPolicy());
      recoveryStats.incrementTime();
    } else if (recoveryStats.isPolicyExceeded()) {
      recoveryFailed(lastShutdownSignal);
      if (!viaConnectionRecovery)
        return;
    }

    try {
      notifyRecoveryStarted();
      delegate = callWithRetries(new Callable<Channel>() {
        @Override
        public Channel call() throws Exception {
          log.info("Recovering {}", ChannelHandler.this);
          previousMaxDeliveryTag = maxDeliveryTag;
          Channel channel = connectionHandler.createChannel(delegate.getChannelNumber());
          migrateConfiguration(channel);
          log.info("Recovered {}", ChannelHandler.this);
          return channel;
        }
      }, config.getChannelRecoveryPolicy(), recoveryStats, config.getRecoverableExceptions(), true,
        false);
      notifyRecovery();
      recoverConsumers(!viaConnectionRecovery);
      recoverySucceeded();
    } catch (Exception e) {
      ShutdownSignalException sse = Exceptions.extractCause(e, ShutdownSignalException.class);
      if (sse != null) {
        if (Exceptions.isConnectionClosure(sse))
          throw e;
      } else if (recoveryStats.isPolicyExceeded())
        recoveryFailed(e);
    }
  }
816910
@Override
	public void channelRead0(ChannelHandlerContext ctx, RoutedRequest routedRequest) throws Exception {
		String requestPath = routedRequest.getPath();

		respondWithFile(ctx, routedRequest.getRequest(), requestPath);
	}
1039648
private void switchChannelStatus(final Long channelId, final ChannelStatus channelStatus) {
        transactionTemplate.execute(new TransactionCallbackWithoutResult() {

            @Override
            protected void doInTransactionWithoutResult(TransactionStatus status) {
                try {
                    final ChannelDO channelDo = channelDao.findById(channelId);

                    if (null == channelDo) {
                        String exceptionCause = "query channelId:" + channelId + " return null.";
                        logger.error("ERROR ## " + exceptionCause);
                        throw new ManagerException(exceptionCause);
                    }

                    ChannelStatus oldStatus = arbitrateManageService.channelEvent().status(channelDo.getId());
                    Channel channel = doToModel(channelDo);
                    // 检查下ddl/home配置
                    List<Pipeline> pipelines = channel.getPipelines();
                    if (pipelines.size() > 1) {
                        boolean ddlSync = true;
                        boolean homeSync = true;
                        for (Pipeline pipeline : pipelines) {
                            homeSync &= pipeline.getParameters().isHome();
                            ddlSync &= pipeline.getParameters().getDdlSync();
                        }

                        if (ddlSync) {
                            throw new InvalidConfigureException(INVALID_TYPE.DDL);
                        }

                        if (homeSync) {
                            throw new InvalidConfigureException(INVALID_TYPE.HOME);
                        }
                    }

                    channel.setStatus(oldStatus);
                    ChannelStatus newStatus = channelStatus;
                    if (newStatus != null) {
                        if (newStatus.equals(oldStatus)) {
                            // String exceptionCause = "switch the channel(" +
                            // channelId + ") status to " +
                            // channelStatus
                            // + " but it had the status:" + oldStatus;
                            // logger.error("ERROR ## " + exceptionCause);
                            // throw new ManagerException(exceptionCause);
                            // ignored
                            return;
                        } else {
                            channel.setStatus(newStatus);// 强制修改为当前变更状态
                        }
                    } else {
                        newStatus = oldStatus;
                    }

                    // 针对关闭操作，要优先更改对应的status，避免node工作线程继续往下跑
                    if (newStatus.isStop()) {
                        arbitrateManageService.channelEvent().stop(channelId);
                    } else if (newStatus.isPause()) {
                        arbitrateManageService.channelEvent().pause(channelId);
                    }

                    // 通知变更内容
                    boolean result = configRemoteService.notifyChannel(channel);// 客户端响应成功，才更改对应的状态

                    if (result) {
                        // 针对启动的话，需要先通知到客户端，客户端启动线程后，再更改channel状态
                        if (newStatus.isStart()) {
                            arbitrateManageService.channelEvent().start(channelId);
                        }
                    }

                } catch (Exception e) {
                    logger.error("ERROR ## switch the channel(" + channelId + ") status has an exception.");
                    throw new ManagerException(e);
                }
            }
        });

    }
510516
public void init(VirtualConnection inVC, HttpOutboundChannel channel) {
        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
            Tr.debug(tc, "Init on link: " + this + " " + inVC);
        }
        super.init(inVC);
        this.myChannel = channel;
        this.bIsActive = true;
        setEnableReconnect(this.myChannel.getHttpConfig().allowsRetries());
        setAllowReconnect(true);
    }
328831
@SuppressWarnings("PMD.DataflowAnomalyAnalysis")
    public void invoke(EventBase<?> event) throws Throwable {
        // ADAPT VERBOSEHANDLERREFERENCE TO ANY CHANGES MADE HERE
        switch (method.type().parameterCount()) {
        case 0:
            // No parameters
            method.invoke();
            break;

        case 1:
            // Event parameter
            method.invoke(event);
            break;

        case 2:
            // Event and channel
            Class<?> channelParam = method.type().parameterType(1);
            for (Channel channel : event.channels()) {
                if (channelParam.isAssignableFrom(channel.getClass())) {
                    method.invoke(event, channel);
                }
            }
            break;

        default:
            throw new IllegalStateException("Handle not usable");
        }
    }