Spring Boot Starter Data Couchbase Reactive

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-couchbase-reactive</artifactId>
</dependency>

implementation 'org.springframework.boot:spring-boot-starter-data-couchbase-reactive'

// Reactive template for programmatic database access
import org.springframework.data.couchbase.core.ReactiveCouchbaseTemplate;

// Base reactive repository interface
import org.springframework.data.couchbase.repository.ReactiveCouchbaseRepository;

// Configuration properties
import org.springframework.boot.autoconfigure.couchbase.CouchbaseProperties;
import org.springframework.boot.autoconfigure.data.couchbase.CouchbaseDataProperties;

// Document mapping annotations
import org.springframework.data.couchbase.core.mapping.Document;
import org.springframework.data.couchbase.core.mapping.Field;
import org.springframework.data.annotation.Id;

// Reactive types
import reactor.core.publisher.Mono;
import reactor.core.publisher.Flux;

// Test support
import org.springframework.boot.test.autoconfigure.data.couchbase.DataCouchbaseTest;

spring:
  couchbase:
    connection-string: couchbase://localhost
    username: admin
    password: password
  data:
    couchbase:
      bucket-name: my-bucket
      auto-index: true

import org.springframework.data.couchbase.core.mapping.Document;
import org.springframework.data.couchbase.core.mapping.Field;
import org.springframework.data.annotation.Id;

/**
 * Couchbase document entity
 */
@Document
public class User {
    
    @Id
    private String id;
    
    @Field
    private String name;
    
    @Field  
    private String email;
    
    // Constructors, getters, setters
    public User() {}
    
    public User(String name, String email) {
        this.name = name;
        this.email = email;
    }
    
    public String getId() { return id; }
    public void setId(String id) { this.id = id; }
    
    public String getName() { return name; }
    public void setName(String name) { this.name = name; }
    
    public String getEmail() { return email; }
    public void setEmail(String email) { this.email = email; }
}

import org.springframework.data.couchbase.repository.ReactiveCouchbaseRepository;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

/**
 * Reactive repository interface for User documents
 * Provides reactive CRUD operations and custom query methods
 */
public interface UserRepository extends ReactiveCouchbaseRepository<User, String> {
    
    /**
     * Find users by name reactively
     * @param name the name to search for
     * @return Flux of matching users
     */
    Flux<User> findByName(String name);
    
    /**
     * Find user by email reactively  
     * @param email the email to search for
     * @return Mono of matching user or empty
     */
    Mono<User> findByEmail(String email);
    
    /**
     * Check if user exists by email
     * @param email the email to check
     * @return Mono of boolean indicating existence
     */
    Mono<Boolean> existsByEmail(String email);
}

import org.springframework.data.couchbase.core.ReactiveCouchbaseTemplate;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Mono;
import reactor.core.publisher.Flux;

/**
 * Service demonstrating reactive template usage
 */
@Service
public class UserService {
    
    private final ReactiveCouchbaseTemplate reactiveCouchbaseTemplate;
    
    public UserService(ReactiveCouchbaseTemplate reactiveCouchbaseTemplate) {
        this.reactiveCouchbaseTemplate = reactiveCouchbaseTemplate;
    }
    
    /**
     * Save user document reactively
     * @param user the user to save
     * @return Mono of saved user
     */
    public Mono<User> saveUser(User user) {
        return reactiveCouchbaseTemplate.save(user);
    }
    
    /**
     * Find user by ID reactively
     * @param id the user ID
     * @return Mono of user or empty
     */
    public Mono<User> findUserById(String id) {
        return reactiveCouchbaseTemplate.findById(User.class).one(id);
    }
    
    /**
     * Find all users reactively
     * @return Flux of all users
     */
    public Flux<User> findAllUsers() {
        return reactiveCouchbaseTemplate.findAll(User.class);
    }
    
    /**
     * Delete user by ID reactively
     * @param id the user ID to delete
     * @return Mono of void indicating completion
     */
    public Mono<Void> deleteUser(String id) {
        return reactiveCouchbaseTemplate.removeById().one(id).then();
    }
}

/**
 * Primary Couchbase connection configuration
 */
public class CouchbaseProperties {
    
    /**
     * Connection string used to locate the Couchbase cluster
     * Example: "couchbase://localhost" or "couchbase://node1,node2"
     */
    private String connectionString;
    
    /**
     * Cluster username for authentication
     */
    private String username;
    
    /**
     * Cluster password for authentication  
     */
    private String password;
    
    /**
     * Environment configuration for fine-tuning
     */
    private Env env = new Env();
}

public class Io {
    /**
     * Minimum number of sockets per node (default: 1)
     */
    private int minEndpoints = 1;
    
    /**
     * Maximum number of sockets per node (default: 12)  
     */
    private int maxEndpoints = 12;
    
    /**
     * HTTP connection idle timeout (default: 4500ms)
     */
    private Duration idleHttpConnectionTimeout = Duration.ofMillis(4500);
}

public class Ssl {
    /**
     * Enable SSL support (auto-detected if keyStore provided)
     */
    private Boolean enabled;
    
    /**
     * Path to JVM key store with certificates
     */
    private String keyStore;
    
    /**
     * Password for key store access
     */
    private String keyStorePassword;
}

public class Timeouts {
    /**
     * Bucket connect timeout (default: 10s)
     */
    private Duration connect = Duration.ofSeconds(10);
    
    /**
     * Bucket disconnect timeout (default: 10s)
     */
    private Duration disconnect = Duration.ofSeconds(10);
    
    /**
     * Key-value operations timeout (default: 2500ms)
     */
    private Duration keyValue = Duration.ofMillis(2500);
    
    /**
     * Durable key-value operations timeout (default: 10s)
     */
    private Duration keyValueDurable = Duration.ofSeconds(10);
    
    /**
     * N1QL query operations timeout (default: 75s)
     */
    private Duration query = Duration.ofSeconds(75);
    
    /**
     * View operations timeout (default: 75s)
     */
    private Duration view = Duration.ofSeconds(75);
    
    /**
     * Search service timeout (default: 75s)
     */
    private Duration search = Duration.ofSeconds(75);
    
    /**
     * Analytics service timeout (default: 75s)
     */
    private Duration analytics = Duration.ofSeconds(75);
    
    /**
     * Management operations timeout (default: 75s)
     */
    private Duration management = Duration.ofSeconds(75);
}

/**
 * Spring Data Couchbase specific configuration
 */
public class CouchbaseDataProperties {
    
    /**
     * Automatically create views and indexes using annotations
     * Uses @ViewIndexed, @N1qlPrimaryIndexed, @N1qlSecondaryIndexed
     */
    private boolean autoIndex = false;
    
    /**
     * Name of the bucket to connect to
     */
    private String bucketName;
    
    /**
     * Name of the scope for collection access
     */
    private String scopeName;
    
    /**
     * Field storing type information for complex types (default: "_class")
     */
    private String typeKey = "_class";
    
    /**
     * Fully qualified FieldNamingStrategy class name for field name mapping
     */
    private Class<?> fieldNamingStrategy;
}

/**
 * ReactiveCouchbaseTemplate provides programmatic database access
 */
public class ReactiveCouchbaseTemplate {
    
    /**
     * Save a document reactively
     * @param entity the entity to save
     * @return Mono of saved entity
     */
    public <T> Mono<T> save(T entity);
    
    /**
     * Find document by ID reactively
     * @param entityClass the entity class
     * @return FindByIdOperation for method chaining
     */
    public <T> FindByIdOperation<T> findById(Class<T> entityClass);
    
    /**
     * Find all documents of type reactively
     * @param entityClass the entity class
     * @return Flux of all entities
     */
    public <T> Flux<T> findAll(Class<T> entityClass);
    
    /**
     * Remove document by ID reactively
     * @return RemoveByIdOperation for method chaining
     */
    public RemoveByIdOperation removeById();
}

/**
 * Advanced reactive query operations
 */
public interface ReactiveQueryOperations {
    
    /**
     * Execute N1QL query reactively
     * @param query the N1QL query string
     * @param entityClass the result entity class
     * @return Flux of query results
     */
    public <T> Flux<T> findByQuery(String query, Class<T> entityClass);
    
    /**
     * Execute view query reactively
     * @param viewName the view name
     * @param entityClass the result entity class  
     * @return Flux of view results
     */
    public <T> Flux<T> findByView(String viewName, Class<T> entityClass);
}

/**
 * Base reactive repository interface providing standard CRUD operations
 * @param <T> the domain type
 * @param <ID> the ID type
 */
public interface ReactiveCouchbaseRepository<T, ID> {
    
    /**
     * Save entity reactively
     * @param entity the entity to save
     * @return Mono of saved entity
     */
    <S extends T> Mono<S> save(S entity);
    
    /**
     * Save multiple entities reactively
     * @param entities the entities to save
     * @return Flux of saved entities
     */
    <S extends T> Flux<S> saveAll(Iterable<S> entities);
    
    /**
     * Find entity by ID reactively
     * @param id the entity ID
     * @return Mono of entity or empty
     */
    Mono<T> findById(ID id);
    
    /**
     * Check if entity exists by ID
     * @param id the entity ID
     * @return Mono of boolean indicating existence
     */
    Mono<Boolean> existsById(ID id);
    
    /**
     * Find all entities reactively
     * @return Flux of all entities
     */
    Flux<T> findAll();
    
    /**
     * Count all entities reactively
     * @return Mono of total count
     */
    Mono<Long> count();
    
    /**
     * Delete entity by ID reactively
     * @param id the entity ID
     * @return Mono of void indicating completion
     */
    Mono<Void> deleteById(ID id);
    
    /**
     * Delete entity reactively
     * @param entity the entity to delete
     * @return Mono of void indicating completion
     */
    Mono<Void> delete(T entity);
    
    /**
     * Delete all entities reactively
     * @return Mono of void indicating completion
     */
    Mono<Void> deleteAll();
}

/**
 * Key beans automatically configured by the starter
 */

/**
 * Main reactive template bean for database operations
 * Bean name: "reactiveCouchbaseTemplate"
 */
@Bean(name = BeanNames.REACTIVE_COUCHBASE_TEMPLATE)
public ReactiveCouchbaseTemplate reactiveCouchbaseTemplate(
    CouchbaseClientFactory couchbaseClientFactory,
    MappingCouchbaseConverter mappingCouchbaseConverter);

/**
 * Repository operations mapping for reactive repositories
 * Bean name: "reactiveCouchbaseRepositoryOperationsMapping"  
 */
@Bean(name = BeanNames.REACTIVE_COUCHBASE_OPERATIONS_MAPPING)
public ReactiveRepositoryOperationsMapping reactiveCouchbaseRepositoryOperationsMapping(
    ReactiveCouchbaseTemplate reactiveCouchbaseTemplate);

/**
 * Cluster environment with I/O, SSL, and timeout configuration
 * Bean name: "couchbaseClusterEnvironment"
 */
@Bean(destroyMethod = "shutdown")
public ClusterEnvironment couchbaseClusterEnvironment(
    CouchbaseProperties properties);

/**
 * Couchbase cluster client connection
 * Bean name: "couchbaseCluster"
 */
@Bean(destroyMethod = "disconnect")
public Cluster couchbaseCluster(
    ClusterEnvironment clusterEnvironment,
    CouchbaseProperties properties);

/**
 * Client factory for bucket access
 * Bean name: "couchbaseClientFactory"
 */
@Bean
public CouchbaseClientFactory couchbaseClientFactory(
    Cluster couchbaseCluster,
    CouchbaseDataProperties properties);

/**
 * Object mapping converter for document serialization
 * Bean name: "couchbaseMappingConverter"
 */
@Bean
public MappingCouchbaseConverter couchbaseMappingConverter(
    CouchbaseMappingContext mappingContext,
    CouchbaseCustomConversions customConversions);

/**
 * Custom type conversions for reactive operations
 * Bean name: "couchbaseCustomConversions"
 */
@Bean
public CouchbaseCustomConversions couchbaseCustomConversions();

/**
 * JSR-303 validation event listener for reactive operations
 * Bean name: "validatingCouchbaseEventListener"
 */
@Bean
public ValidatingCouchbaseEventListener validationEventListener();

/**
 * Standard bean name constants for Couchbase components
 */
public interface BeanNames {
    String COUCHBASE_TEMPLATE = "couchbaseTemplate";
    String REACTIVE_COUCHBASE_TEMPLATE = "reactiveCouchbaseTemplate";
    String COUCHBASE_MAPPING_CONTEXT = "couchbaseMappingContext";
    String COUCHBASE_CUSTOM_CONVERSIONS = "couchbaseCustomConversions";
    String COUCHBASE_OPERATIONS_MAPPING = "couchbaseOperationsMapping";
    String REACTIVE_COUCHBASE_OPERATIONS_MAPPING = "reactiveCouchbaseRepositoryOperationsMapping";
}

/**
 * Interface for customizing cluster environment before cluster creation
 * Implement this interface and register as a bean to customize advanced settings
 */
@FunctionalInterface
public interface ClusterEnvironmentBuilderCustomizer {
    
    /**
     * Customize the cluster environment builder
     * @param builder the builder to customize
     */
    void customize(ClusterEnvironment.Builder builder);
}

/**
 * Interface for registering custom type conversions
 */
public class CouchbaseCustomConversions {
    
    /**
     * Create with custom converters for reactive operations
     * @param converters list of custom converters
     */
    public CouchbaseCustomConversions(List<?> converters);
    
    /**
     * Create with default reactive-compatible converters
     */
    public CouchbaseCustomConversions();
}

/**
 * Base interface for custom converters
 */
public interface Converter<S, T> {
    T convert(S source);
}

/**
 * Event listener providing JSR-303 validation for reactive operations
 * Automatically validates entities before save operations
 */
public class ValidatingCouchbaseEventListener {
    
    /**
     * Validates entity before save operations
     * @param source the entity being saved
     */
    @EventListener
    public void onBeforeSave(BeforeSaveEvent<Object> source);
}

/**
 * Test annotation for Couchbase data layer testing
 * Configures only Couchbase-related components for fast testing
 */
@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Inherited
@BootstrapWith(DataCouchbaseTestContextBootstrapper.class)
@ExtendWith(SpringExtension.class)
@OverrideAutoConfiguration(enabled = false)
@TypeExcludeFilters(DataCouchbaseTypeExcludeFilter.class)
@AutoConfigureCache
@AutoConfigureDataCouchbase
@AutoConfigureTestDatabase
@ImportAutoConfiguration
public @interface DataCouchbaseTest {
    
    /**
     * Properties to add to test environment in key=value format
     * @return array of key=value properties
     */
    String[] properties() default {};
    
    /**
     * Whether to use default filters for component scanning (default: true)
     * @return true to use default type filters
     */
    boolean useDefaultFilters() default true;
    
    /**
     * Additional include filters for component scanning
     * @return include filters array
     */
    Filter[] includeFilters() default {};
    
    /**
     * Additional exclude filters for component scanning  
     * @return exclude filters array
     */
    Filter[] excludeFilters() default {};
    
    /**
     * Auto-configuration exclusions for this test
     * @return classes to exclude from auto-configuration
     */
    Class<?>[] excludeAutoConfiguration() default {};
    
    /**
     * Component filter for fine-grained test configuration
     */
    @Target({})
    @Retention(RetentionPolicy.RUNTIME)
    @Documented
    @interface Filter {
        FilterType type() default FilterType.ANNOTATION;
        Class<?>[] classes() default {};
        String[] pattern() default {};
    }
}

/**
 * Auto-configuration annotation for manual Couchbase setup in tests
 * Alternative to @DataCouchbaseTest for custom test configurations
 */
@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@Documented
@ImportAutoConfiguration
public @interface AutoConfigureDataCouchbase {
}

/**
 * Test context bootstrapper for Couchbase data tests
 * Customizes Spring test context for Couchbase-only components
 */
public class DataCouchbaseTestContextBootstrapper extends SpringBootTestContextBootstrapper {
}

/**
 * Type exclude filter for Couchbase data tests
 * Excludes non-Couchbase components from test context
 */
public class DataCouchbaseTypeExcludeFilter extends AnnotationCustomizableTypeExcludeFilter {
}

@DataCouchbaseTest
class UserRepositoryTest {
    
    @Autowired
    private TestEntityManager entityManager;
    
    @Autowired
    private UserRepository userRepository;
    
    @Test
    void shouldFindUserByEmail() {
        // Given
        User user = new User("John Doe", "john@example.com");
        entityManager.save(user);
        
        // When
        StepVerifier.create(userRepository.findByEmail("john@example.com"))
            // Then
            .expectNextMatches(found -> "John Doe".equals(found.getName()))
            .verifyComplete();
    }
}

/**
 * Common exceptions thrown by reactive Couchbase operations
 */

/**
 * Thrown when document is not found
 */
public class DocumentNotFoundException extends CouchbaseException;

/**
 * Thrown when bucket is not available
 */
public class BucketNotFoundException extends CouchbaseException;

/**
 * Thrown on authentication failures
 */
public class AuthenticationFailureException extends CouchbaseException;

/**
 * Thrown on connection timeouts
 */
public class TimeoutException extends CouchbaseException;

// Handle errors in reactive chains
userRepository.findById("user123")
    .switchIfEmpty(Mono.error(new UserNotFoundException("User not found")))
    .onErrorResume(TimeoutException.class, ex -> 
        Mono.error(new ServiceUnavailableException("Database timeout")))
    .subscribe(
        user -> log.info("Found user: {}", user.getName()),
        error -> log.error("Error finding user", error)
    );

/**
 * Configure multiple buckets by defining custom CouchbaseClientFactory beans
 */
@Configuration
public class MultiCouchbaseConfig {
    
    @Bean
    @Primary
    public CouchbaseClientFactory primaryBucketFactory(
        Cluster couchbaseCluster,
        CouchbaseDataProperties properties) {
        return new SimpleCouchbaseClientFactory(couchbaseCluster, 
            properties.getBucketName(), properties.getScopeName());
    }
    
    @Bean
    public CouchbaseClientFactory secondaryBucketFactory(
        Cluster couchbaseCluster) {
        return new SimpleCouchbaseClientFactory(couchbaseCluster, 
            "secondary-bucket", null);
    }
    
    @Bean
    public ReactiveCouchbaseTemplate secondaryTemplate(
        @Qualifier("secondaryBucketFactory") CouchbaseClientFactory factory,
        MappingCouchbaseConverter converter) {
        return new ReactiveCouchbaseTemplate(factory, converter);
    }
}

/**
 * Configure custom field naming for document mapping
 */
@Configuration
public class CouchbaseFieldNamingConfig {
    
    @Bean
    public PropertyNamingStrategy fieldNamingStrategy() {
        return PropertyNamingStrategies.SNAKE_CASE;
    }
}

/**
 * Example of advanced cluster environment customization
 */
@Configuration
public class CustomClusterConfig {
    
    @Bean
    public ClusterEnvironmentBuilderCustomizer clusterCustomizer() {
        return builder -> {
            // Custom JSON serializer
            builder.jsonSerializer(JacksonJsonSerializer.create());
            
            // Advanced I/O configuration
            builder.ioConfig(io -> io
                .numKvConnections(8)
                .networkResolution(NetworkResolution.AUTO)
                .enableMutationTokens(false));
                
            // Security configuration
            builder.securityConfig(security -> security
                .enableTls(true)
                .enableHostnameVerification(true)
                .trustManagerFactory(getTrustManagerFactory()));
                
            // Compression configuration  
            builder.compressionConfig(compression -> compression
                .enable(true)
                .minSize(32)
                .minRatio(0.83f));
        };
    }
}