begin_unit|revision:0.9.5;language:Java;cregit-version:0.0.1
begin_comment
comment|/*  * 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.  */
end_comment

begin_package
DECL|package|org.apache.camel.component.reactive.streams
package|package
name|org
operator|.
name|apache
operator|.
name|camel
operator|.
name|component
operator|.
name|reactive
operator|.
name|streams
package|;
end_package

begin_import
import|import
name|java
operator|.
name|util
operator|.
name|LinkedList
import|;
end_import

begin_import
import|import
name|java
operator|.
name|util
operator|.
name|List
import|;
end_import

begin_import
import|import
name|java
operator|.
name|util
operator|.
name|concurrent
operator|.
name|CountDownLatch
import|;
end_import

begin_import
import|import
name|java
operator|.
name|util
operator|.
name|concurrent
operator|.
name|TimeUnit
import|;
end_import

begin_import
import|import
name|io
operator|.
name|reactivex
operator|.
name|Observable
import|;
end_import

begin_import
import|import
name|org
operator|.
name|apache
operator|.
name|camel
operator|.
name|Exchange
import|;
end_import

begin_import
import|import
name|org
operator|.
name|apache
operator|.
name|camel
operator|.
name|RoutesBuilder
import|;
end_import

begin_import
import|import
name|org
operator|.
name|apache
operator|.
name|camel
operator|.
name|builder
operator|.
name|RouteBuilder
import|;
end_import

begin_import
import|import
name|org
operator|.
name|apache
operator|.
name|camel
operator|.
name|component
operator|.
name|reactive
operator|.
name|streams
operator|.
name|api
operator|.
name|CamelReactiveStreams
import|;
end_import

begin_import
import|import
name|org
operator|.
name|apache
operator|.
name|camel
operator|.
name|test
operator|.
name|junit4
operator|.
name|CamelTestSupport
import|;
end_import

begin_import
import|import
name|org
operator|.
name|junit
operator|.
name|Test
import|;
end_import

begin_class
DECL|class|PublisherTypeConversionTest
specifier|public
class|class
name|PublisherTypeConversionTest
extends|extends
name|CamelTestSupport
block|{
annotation|@
name|Test
DECL|method|testConversion ()
specifier|public
name|void
name|testConversion
parameter_list|()
throws|throws
name|Exception
block|{
name|CountDownLatch
name|latch
init|=
operator|new
name|CountDownLatch
argument_list|(
literal|3
argument_list|)
decl_stmt|;
name|List
argument_list|<
name|Integer
argument_list|>
name|integers
init|=
operator|new
name|LinkedList
argument_list|<>
argument_list|()
decl_stmt|;
name|Observable
operator|.
name|fromPublisher
argument_list|(
name|CamelReactiveStreams
operator|.
name|get
argument_list|(
name|context
argument_list|)
operator|.
name|fromStream
argument_list|(
literal|"pub"
argument_list|,
name|Exchange
operator|.
name|class
argument_list|)
argument_list|)
operator|.
name|map
argument_list|(
name|x
lambda|->
name|x
operator|.
name|getIn
argument_list|()
operator|.
name|getBody
argument_list|(
name|Integer
operator|.
name|class
argument_list|)
argument_list|)
operator|.
name|subscribe
argument_list|(
name|n
lambda|->
block|{
name|integers
operator|.
name|add
argument_list|(
name|n
argument_list|)
argument_list|;
name|latch
operator|.
name|countDown
argument_list|()
argument_list|;
block|}
block|)
class|;
end_class

begin_expr_stmt
name|Observable
operator|.
name|fromPublisher
argument_list|(
name|CamelReactiveStreams
operator|.
name|get
argument_list|(
name|context
argument_list|)
operator|.
name|fromStream
argument_list|(
literal|"pub"
argument_list|)
argument_list|)
operator|.
name|map
argument_list|(
name|x
lambda|->
name|x
operator|.
name|getIn
argument_list|()
operator|.
name|getBody
argument_list|(
name|Integer
operator|.
name|class
argument_list|)
argument_list|)
operator|.
name|subscribe
argument_list|(
name|n
lambda|->
block|{
name|integers
operator|.
name|add
argument_list|(
name|n
argument_list|)
argument_list|;
name|latch
operator|.
name|countDown
argument_list|()
argument_list|;
end_expr_stmt

begin_empty_stmt
unit|})
empty_stmt|;
end_empty_stmt

begin_expr_stmt
name|Observable
operator|.
name|fromPublisher
argument_list|(
name|CamelReactiveStreams
operator|.
name|get
argument_list|(
name|context
argument_list|)
operator|.
name|fromStream
argument_list|(
literal|"pub"
argument_list|,
name|Integer
operator|.
name|class
argument_list|)
argument_list|)
operator|.
name|subscribe
argument_list|(
name|n
lambda|->
block|{
name|integers
operator|.
name|add
argument_list|(
name|n
argument_list|)
expr_stmt|;
name|latch
operator|.
name|countDown
argument_list|()
expr_stmt|;
block|}
argument_list|)
expr_stmt|;
end_expr_stmt

begin_expr_stmt
name|context
operator|.
name|start
argument_list|()
expr_stmt|;
end_expr_stmt

begin_expr_stmt
name|latch
operator|.
name|await
argument_list|(
literal|5
argument_list|,
name|TimeUnit
operator|.
name|SECONDS
argument_list|)
expr_stmt|;
end_expr_stmt

begin_expr_stmt
name|assertEquals
argument_list|(
literal|3
argument_list|,
name|integers
operator|.
name|size
argument_list|()
argument_list|)
expr_stmt|;
end_expr_stmt

begin_for
for|for
control|(
name|int
name|i
range|:
name|integers
control|)
block|{
name|assertEquals
argument_list|(
literal|123
argument_list|,
name|i
argument_list|)
expr_stmt|;
block|}
end_for

begin_function
unit|}      @
name|Override
DECL|method|createRouteBuilder ()
specifier|protected
name|RoutesBuilder
name|createRouteBuilder
parameter_list|()
throws|throws
name|Exception
block|{
return|return
operator|new
name|RouteBuilder
argument_list|()
block|{
annotation|@
name|Override
specifier|public
name|void
name|configure
parameter_list|()
throws|throws
name|Exception
block|{
name|from
argument_list|(
literal|"timer:tick?period=50&repeatCount=1"
argument_list|)
operator|.
name|setBody
argument_list|()
operator|.
name|constant
argument_list|(
literal|123
argument_list|)
operator|.
name|to
argument_list|(
literal|"reactive-streams:pub"
argument_list|)
expr_stmt|;
block|}
block|}
return|;
block|}
end_function

begin_function
annotation|@
name|Override
DECL|method|isUseAdviceWith ()
specifier|public
name|boolean
name|isUseAdviceWith
parameter_list|()
block|{
return|return
literal|true
return|;
block|}
end_function

unit|}
end_unit

