本文整理汇总了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；未经允许，请勿转载。 |
客服电话
APP下载
官方微信
请发表评论