davsclaus commented on code in PR #25034: URL: https://github.com/apache/camel/pull/25034#discussion_r3635875549
########## components/camel-clickhouse/src/main/java/org/apache/camel/component/clickhouse/ClickHouseEndpoint.java: ########## @@ -0,0 +1,259 @@ +/* + * 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. + */ +package org.apache.camel.component.clickhouse; + +import com.clickhouse.client.api.Client; +import org.apache.camel.Category; +import org.apache.camel.Consumer; +import org.apache.camel.Processor; +import org.apache.camel.Producer; +import org.apache.camel.spi.Metadata; +import org.apache.camel.spi.UriEndpoint; +import org.apache.camel.spi.UriParam; +import org.apache.camel.spi.UriPath; +import org.apache.camel.support.DefaultEndpoint; +import org.apache.camel.util.ObjectHelper; + +/** + * Interact with <a href="https://clickhouse.com/";>ClickHouse</a>, the high-performance columnar OLAP database, for + * high-throughput ingestion and OLAP queries. + */ +@UriEndpoint(firstVersion = "4.22.0", scheme = "clickhouse", title = "ClickHouse", + syntax = "clickhouse:database", category = { Category.DATABASE, Category.BIGDATA }, + producerOnly = true, headersClass = ClickHouseConstants.class) +public class ClickHouseEndpoint extends DefaultEndpoint { + + private volatile Client client; + private boolean clientCreated; + + @UriPath + @Metadata(required = true, + description = "The ClickHouse database. A table may also be provided using the database.table syntax.") + private String database; + @UriParam(description = "The target table for insert operations. Can also be given in the path as database.table" + + " or overridden per message with the CamelClickHouseTable header.") + private String table; + @UriParam(description = "The ClickHouse HTTP endpoint URL, e.g. http://localhost:8123. Required unless a shared" + + " Client bean is autowired or configured on the component.") + private String serverUrl; + @UriParam(label = "security", defaultValue = "default", description = "The username used to authenticate to ClickHouse.") + private String username = "default"; + @UriParam(label = "security", secret = true, description = "The password used to authenticate to ClickHouse.") + private String password; + @UriParam(label = "security", defaultValue = "false", + description = "Whether to connect to ClickHouse over a secure (HTTPS) connection.") + private boolean ssl; + @UriParam(defaultValue = "INSERT", description = "The operation to perform: insert, query or ping.") + private ClickHouseOperation operation = ClickHouseOperation.INSERT; + @UriParam(defaultValue = "JSONEachRow", + description = "The ClickHouse data format used for insert and query operations, e.g. JSONEachRow," + + " RowBinary, CSV, TSV or Parquet.") + private String format = "JSONEachRow"; + @UriParam(defaultValue = "0", + description = "The client-side batch size for insert operations when the message body is a List. When" + + " greater than 0, the list is split into batches of this size and each batch is sent as a" + + " separate insert. A value of 0 (default) inserts the whole list in a single call.") + private int batchSize; + @UriParam(defaultValue = "false", + description = "Whether to use ClickHouse server-side asynchronous inserts (async_insert=1).") + private boolean asyncInsert; + @UriParam(defaultValue = "true", + description = "When asyncInsert is enabled, whether the server should wait for the async insert to be" + + " flushed before acknowledging (wait_for_async_insert).") + private boolean waitForAsyncInsert = true; + @UriParam(defaultValue = "false", + description = "Whether to compress the insert request payload sent to the server (LZ4).") + private boolean compression; + + public ClickHouseEndpoint() { + } + + public ClickHouseEndpoint(String uri, ClickHouseComponent component) { + super(uri, component); + } + + @Override + public Producer createProducer() throws Exception { + return new ClickHouseProducer(this); + } + + @Override + public Consumer createConsumer(Processor processor) throws Exception { + throw new UnsupportedOperationException("You cannot receive messages from this endpoint"); + } + + @Override + protected void doStart() throws Exception { + super.doStart(); + if (client == null && ObjectHelper.isEmpty(serverUrl)) { + throw new IllegalArgumentException( + "Either a shared Client must be autowired or the serverUrl option must be configured"); + } + } + + @Override + protected void doStop() throws Exception { + if (clientCreated && client != null) { + client.close(); + client = null; + clientCreated = false; + } + super.doStop(); + } + + /** + * Returns the ClickHouse client, lazily building one from the endpoint connection options (serverUrl, username, + * password, ...) when no shared client has been provided. + */ + public Client getClient() { + if (client == null && ObjectHelper.isNotEmpty(serverUrl)) { + synchronized (this) { Review Comment: do not create lazy but instead build in doStart if no client was pre configured - this is the standard that Camel components do ########## components/camel-clickhouse/src/main/java/org/apache/camel/component/clickhouse/ClickHouseProducer.java: ########## @@ -0,0 +1,202 @@ +/* + * 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. + */ +package org.apache.camel.component.clickhouse; + +import java.io.ByteArrayInputStream; +import java.io.File; +import java.io.FileInputStream; +import java.io.InputStream; +import java.nio.charset.StandardCharsets; +import java.util.List; + +import com.clickhouse.client.api.Client; +import com.clickhouse.client.api.insert.InsertResponse; +import com.clickhouse.client.api.insert.InsertSettings; +import com.clickhouse.client.api.query.QueryResponse; +import com.clickhouse.client.api.query.QuerySettings; +import com.clickhouse.data.ClickHouseFormat; +import org.apache.camel.Exchange; +import org.apache.camel.Message; +import org.apache.camel.support.DefaultProducer; +import org.apache.camel.util.IOHelper; +import org.apache.camel.util.ObjectHelper; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class ClickHouseProducer extends DefaultProducer { + + private static final Logger LOG = LoggerFactory.getLogger(ClickHouseProducer.class); + + private final ClickHouseEndpoint endpoint; + + public ClickHouseProducer(ClickHouseEndpoint endpoint) { + super(endpoint); + this.endpoint = endpoint; + } + + @Override + public void process(Exchange exchange) throws Exception { + ClickHouseOperation operation = resolveOperation(exchange); + switch (operation) { + case INSERT -> doInsert(exchange); + case QUERY -> doQuery(exchange); + case PING -> doPing(exchange); + default -> throw new ClickHouseException("Unsupported operation: " + operation); + } + } + + private void doInsert(Exchange exchange) throws Exception { + Client client = endpoint.getClient(); + String target = resolveTable(exchange); + if (ObjectHelper.isEmpty(target)) { + throw new ClickHouseException( + "A table is required for insert operations. Provide it via the URI path (database.table)," + + " the table option, or the " + ClickHouseConstants.TABLE + " header."); + } + + InsertSettings settings = new InsertSettings(); + String database = resolveDatabase(exchange); + if (ObjectHelper.isNotEmpty(database)) { + settings.setDatabase(database); + } + if (endpoint.isAsyncInsert()) { + settings.serverSetting("async_insert", "1"); + settings.serverSetting("wait_for_async_insert", endpoint.isWaitForAsyncInsert() ? "1" : "0"); + } + + Object body = exchange.getIn().getBody(); + long writtenRows; + if (body instanceof List) { + List<?> data = exchange.getIn().getBody(List.class); + writtenRows = insertList(client, target, data, settings); + } else { + InputStream stream = null; + boolean createdStream = false; + try { + if (body instanceof InputStream is) { + stream = is; + } else if (body instanceof byte[] bytes) { + stream = new ByteArrayInputStream(bytes); + createdStream = true; + } else if (body instanceof File file) { Review Comment: Look at WrappedFile as that is how camel-file etc would provide, you can check other components for this class and what they do -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
