Java Input类代码示例

OStack程序员社区-中国程序员成长平台 › 门户 › 编程› Java›Java编程经验

原作者: [db:作者] 来自: [db:来源] 收藏邀请

本文整理汇总了Java中org.springframework.cloud.stream.annotation.Input类的典型用法代码示例。如果您正苦于以下问题：Java Input类的具体用法？Java Input怎么用？Java Input使用的例子？那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。

Input类属于org.springframework.cloud.stream.annotation包，在下文中一共展示了Input类的20个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Java代码示例。

示例1: receive

import org.springframework.cloud.stream.annotation.Input; //导入依赖的package包/类
@StreamListener
public void receive(@Input(Processor.INPUT) SubscribableChannel input,
		@Output(Processor.OUTPUT) final MessageChannel output1,
		@Output(StreamListenerTestUtils.FooOutboundChannel1.OUTPUT) final MessageChannel output2) {
	input.subscribe(new MessageHandler() {
		@Override
		public void handleMessage(Message<?> message) throws MessagingException {
			if (message.getHeaders().get("output").equals("output1")) {
				output1.send(org.springframework.messaging.support.MessageBuilder
						.withPayload(message.getPayload().toString().toUpperCase()).build());
			}
			else if (message.getHeaders().get("output").equals("output2")) {
				output2.send(org.springframework.messaging.support.MessageBuilder
						.withPayload(message.getPayload().toString().toLowerCase()).build());
			}
		}
	});
}

开发者ID:spring-cloud，项目名称:spring-cloud-stream，代码行数:19，代码来源:StreamListenerHandlerMethodTests.java

示例2: start

import org.springframework.cloud.stream.annotation.Input; //导入依赖的package包/类
@Override
public void start() {
	try {
		lock.lock();
		if (!running) {
			mappedStreamEmitterMethods.forEach((k, v) -> v.forEach(item -> {
						Assert.isTrue(item.getAnnotation(Input.class) == null,
								StreamEmitterErrorMessages.INPUT_ANNOTATIONS_ARE_NOT_ALLOWED);
						String methodAnnotatedOutboundName =
								StreamAnnotationCommonMethodUtils.getOutboundBindingTargetName(item);
						int outputAnnotationCount = StreamAnnotationCommonMethodUtils.outputAnnotationCount(item);
						validateStreamEmitterMethod(item, outputAnnotationCount, methodAnnotatedOutboundName);
						invokeSetupMethodOnToTargetChannel(item, k, methodAnnotatedOutboundName);
					}
			));
			this.running = true;
		}
	}
	finally {
		lock.unlock();
	}
}

开发者ID:spring-cloud，项目名称:spring-cloud-stream，代码行数:23，代码来源:StreamEmitterAnnotationBeanPostProcessor.java

示例3: save

import org.springframework.cloud.stream.annotation.Input; //导入依赖的package包/类
@StreamListener
@Output(CustomProcessor.OUTPUT)
public Flux<Void> save(@Input(CustomProcessor.INPUT)
					   Flux<Comment> newComments) {
	return repository
		.saveAll(newComments)
		.flatMap(comment -> {
			meterRegistry
				.counter("comments.consumed", "imageId", comment.getImageId())
				.increment();
			return Mono.empty();
		});
}

开发者ID:PacktPublishing，项目名称:Learning-Spring-Boot-2.0-Second-Edition，代码行数:14，代码来源:CommentService.java

示例4: save

import org.springframework.cloud.stream.annotation.Input; //导入依赖的package包/类
@StreamListener
@Output(Processor.OUTPUT)
public Flux<Comment> save(@Input(Processor.INPUT) Flux<Comment> newComment) {
	return repository
		.saveAll(newComment)
		.map(comment -> {
			meterRegistry
				.counter("comments.consumed", "imageId", comment.getImageId())
				.increment();
			return comment;
		});
}

开发者ID:PacktPublishing，项目名称:Learning-Spring-Boot-2.0-Second-Edition，代码行数:13，代码来源:CommentService.java

示例5: save

import org.springframework.cloud.stream.annotation.Input; //导入依赖的package包/类
@StreamListener
@Output(Processor.OUTPUT)
public Flux<Comment> save(@Input(Processor.INPUT) Flux<Comment> newComment) {
	return repository
		.saveAll(newComment)
		.map(comment -> {
			log.info("Saving new comment " + comment);
			meterRegistry
				.counter("comments.consumed", "imageId", comment.getImageId())
				.increment();
			return comment;
		});
}

开发者ID:PacktPublishing，项目名称:Learning-Spring-Boot-2.0-Second-Edition，代码行数:14，代码来源:CommentService.java

示例6: save

import org.springframework.cloud.stream.annotation.Input; //导入依赖的package包/类
@StreamListener
@Output(Processor.OUTPUT)
public Flux<Void> save(@Input(Processor.INPUT) Flux<Comment> newComment) {
	return repository
		.saveAll(newComment)
		.flatMap(comment -> {
			meterRegistry
				.counter("comments.consumed", "imageId", comment.getImageId())
				.increment();
			return Mono.empty();
		});
}

开发者ID:PacktPublishing，项目名称:Learning-Spring-Boot-2.0-Second-Edition，代码行数:13，代码来源:CommentService.java

示例7: verifyEmpString

import org.springframework.cloud.stream.annotation.Input; //导入依赖的package包/类
@StreamListener
@Output(Processor.OUTPUT)
public Flux<String> verifyEmpString(@Input(Processor.INPUT) Flux<String> id) {
	System.out.println("first");
	id.delayElements(Duration.ofMillis(2))
	   .log();
	return id;
}

开发者ID:PacktPublishing，项目名称:Spring-5.0-Cookbook，代码行数:9，代码来源:EmpIdConverterConverter.java

示例8: toUpperCase

import org.springframework.cloud.stream.annotation.Input; //导入依赖的package包/类
@StreamListener
@Output(Processor.OUTPUT)
public Flux<String> toUpperCase(@Input(Processor.INPUT) Flux<String> inbound) {
	return inbound.
			log()
			.window(Duration.ofSeconds(10), Duration.ofSeconds(5))
			.flatMap(w -> w.reduce("", (s1,s2)->s1+s2))
			.log();
}

开发者ID:spring-cloud，项目名称:spring-cloud-stream-samples，代码行数:10，代码来源:ReactiveKafkaProcessorApplication.java

示例9: postProcessAfterInitialization

import org.springframework.cloud.stream.annotation.Input; //导入依赖的package包/类
@Override
public final Object postProcessAfterInitialization(Object bean, final String beanName) throws BeansException {
	Class<?> targetClass = AopUtils.isAopProxy(bean) ? AopUtils.getTargetClass(bean) : bean.getClass();
	Method[] uniqueDeclaredMethods = ReflectionUtils.getUniqueDeclaredMethods(targetClass);
	for (Method method : uniqueDeclaredMethods) {
		StreamListener streamListener = AnnotatedElementUtils.findMergedAnnotation(method, StreamListener.class);
		if (streamListener != null && !method.isBridge()) {
			streamListenerCallbacks.add(() -> {
				Assert.isTrue(method.getAnnotation(Input.class) == null, StreamListenerErrorMessages.INPUT_AT_STREAM_LISTENER);
				this.doPostProcess(streamListener, method, bean);
			});
		}
	}
	return bean;
}

开发者ID:spring-cloud，项目名称:spring-cloud-stream，代码行数:16，代码来源:StreamListenerAnnotationBeanPostProcessor.java

示例10: checkDeclarativeMethod

import org.springframework.cloud.stream.annotation.Input; //导入依赖的package包/类
private boolean checkDeclarativeMethod(Method method, String methodAnnotatedInboundName, String methodAnnotatedOutboundName) {
	int methodArgumentsLength = method.getParameterTypes().length;
	for (int parameterIndex = 0; parameterIndex < methodArgumentsLength; parameterIndex++) {
		MethodParameter methodParameter = MethodParameter.forExecutable(method, parameterIndex);
		if (methodParameter.hasParameterAnnotation(Input.class)) {
			String inboundName = (String) AnnotationUtils
					.getValue(methodParameter.getParameterAnnotation(Input.class));
			Assert.isTrue(StringUtils.hasText(inboundName), StreamListenerErrorMessages.INVALID_INBOUND_NAME);
			Assert.isTrue(isDeclarativeMethodParameter(inboundName, methodParameter),
					StreamListenerErrorMessages.INVALID_DECLARATIVE_METHOD_PARAMETERS);
			return true;
		}
		else if (methodParameter.hasParameterAnnotation(Output.class)) {
			String outboundName = (String) AnnotationUtils
					.getValue(methodParameter.getParameterAnnotation(Output.class));
			Assert.isTrue(StringUtils.hasText(outboundName), StreamListenerErrorMessages.INVALID_OUTBOUND_NAME);
			Assert.isTrue(isDeclarativeMethodParameter(outboundName, methodParameter),
					StreamListenerErrorMessages.INVALID_DECLARATIVE_METHOD_PARAMETERS);
			return true;
		}
		else if (StringUtils.hasText(methodAnnotatedOutboundName)) {
			return isDeclarativeMethodParameter(methodAnnotatedOutboundName, methodParameter);
		}
		else if (StringUtils.hasText(methodAnnotatedInboundName)) {
			return isDeclarativeMethodParameter(methodAnnotatedInboundName, methodParameter);
		}
	}
	return false;
}

开发者ID:spring-cloud，项目名称:spring-cloud-stream，代码行数:30，代码来源:StreamListenerAnnotationBeanPostProcessor.java

示例11: receive

import org.springframework.cloud.stream.annotation.Input; //导入依赖的package包/类
@StreamListener
public void receive(
		@Input(Processor.INPUT) @Payload StreamListenerTestUtils.FooPojo fooPojo,
		@Headers Map<String, Object> headers,
		@Header(MessageHeaders.CONTENT_TYPE) String contentType) {
	this.receivedArguments.add(fooPojo);
	this.receivedArguments.add(headers);
	this.receivedArguments.add(contentType);
}

开发者ID:spring-cloud，项目名称:spring-cloud-stream，代码行数:10，代码来源:StreamListenerAnnotatedMethodArgumentsTests.java

示例12: receive

import org.springframework.cloud.stream.annotation.Input; //导入依赖的package包/类
@StreamListener
public void receive(@Input(Processor.INPUT) SubscribableChannel input,
		@Output(Processor.OUTPUT) final MessageChannel output) {
	input.subscribe(new MessageHandler() {
		@Override
		public void handleMessage(Message<?> message) throws MessagingException {
			output.send(MessageBuilder.withPayload(message.getPayload().toString().toUpperCase()).build());
		}
	});
}

开发者ID:spring-cloud，项目名称:spring-cloud-stream，代码行数:11，代码来源:StreamListenerWithAnnotatedInputOutputArgsTests.java

示例13: emit

import org.springframework.cloud.stream.annotation.Input; //导入依赖的package包/类
@StreamEmitter
@Output(Source.OUTPUT)
@Input(Processor.INPUT)
public Flux<String> emit() {
	return Flux.interval(Duration.ofMillis(1))
			.map(l -> "Hello World!!" + l);
}

开发者ID:spring-cloud，项目名称:spring-cloud-stream，代码行数:8，代码来源:StreamEmitterValidationTests.java

示例14: receive

import org.springframework.cloud.stream.annotation.Input; //导入依赖的package包/类
@StreamListener
public void receive(@Input(Sink.INPUT) Flux<String> input) {
	input.window(Duration.ofMillis(500), Duration.ofMillis(100))
			.flatMap(w -> w.reduce("", (x, y) -> x + y))
			.subscribe(x -> {
				interruptionState = Thread.currentThread().isInterrupted();
				latch.countDown();
			});
}

开发者ID:spring-cloud，项目名称:spring-cloud-stream，代码行数:10，代码来源:StreamListenerInterruptionTests.java

示例15: receive

import org.springframework.cloud.stream.annotation.Input; //导入依赖的package包/类
@StreamListener
public void receive(@Input(Processor.INPUT) Flux<Message<String>> input,
		@Output(Processor.OUTPUT) FluxSender output) {
	output.send(input
			.map(m -> m.getPayload().toString())
			.map(m -> {
				if (!m.equals("fail")) {
					return m.toUpperCase();
				}
				else {
					throw new RuntimeException();
				}
			})
			.map(o -> MessageBuilder.withPayload(o).build()));
}

开发者ID:spring-cloud，项目名称:spring-cloud-stream，代码行数:16，代码来源:StreamListenerReactiveInputOutputArgsWithSenderAndFailureTests.java

示例16: receive

import org.springframework.cloud.stream.annotation.Input; //导入依赖的package包/类
@StreamListener
public @Output(Processor.OUTPUT) Flux<String> receive(@Input(Processor.INPUT) Flux<String> input) {
	return input.map(m -> {
		if (!m.equals("fail")) {
			return m.toUpperCase();
		}
		else {
			throw new RuntimeException();
		}
	});
}

开发者ID:spring-cloud，项目名称:spring-cloud-stream，代码行数:12，代码来源:StreamListenerReactiveReturnWithFailureTests.java

示例17: receive

import org.springframework.cloud.stream.annotation.Input; //导入依赖的package包/类
@StreamListener
public void receive(@Input(Processor.INPUT) Flux<Message<String>> input,
		@Output(Processor.OUTPUT) FluxSender output) {
	output.send(input
			.map(m -> m.getPayload().toString().toUpperCase())
			.map(o -> MessageBuilder.withPayload(o).build()));
}

开发者ID:spring-cloud，项目名称:spring-cloud-stream，代码行数:8，代码来源:StreamListenerReactiveInputOutputArgsWithSenderTests.java

示例18: input

import org.springframework.cloud.stream.annotation.Input; //导入依赖的package包/类
@Input(CustomProcessor.INPUT)
SubscribableChannel input();

开发者ID:PacktPublishing，项目名称:Learning-Spring-Boot-2.0-Second-Edition，代码行数:3，代码来源:CustomProcessor.java

示例19: newComments

import org.springframework.cloud.stream.annotation.Input; //导入依赖的package包/类
@Input(NEW_COMMENTS)
SubscribableChannel newComments();

开发者ID:PacktPublishing，项目名称:Learning-Spring-Boot-2.0-Second-Edition，代码行数:3，代码来源:ChatServiceStreams.java

示例20: brokerToClient

import org.springframework.cloud.stream.annotation.Input; //导入依赖的package包/类
@Input(BROKER_TO_CLIENT)
SubscribableChannel brokerToClient();

开发者ID:PacktPublishing，项目名称:Learning-Spring-Boot-2.0-Second-Edition，代码行数:3，代码来源:ChatServiceStreams.java

注：本文中的org.springframework.cloud.stream.annotation.Input类示例整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。