-
Notifications
You must be signed in to change notification settings - Fork 30
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
API gateway: cache topic producer for the same gateway
- Loading branch information
1 parent
7897c34
commit 5f38cd8
Showing
15 changed files
with
314 additions
and
31 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
37 changes: 37 additions & 0 deletions
37
langstream-api-gateway/src/main/java/ai/langstream/apigateway/config/TopicProperties.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,37 @@ | ||
| /* | ||
| * Copyright DataStax, Inc. | ||
| * | ||
| * Licensed 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 ai.langstream.apigateway.config; | ||
|
|
||
| import com.fasterxml.jackson.annotation.JsonProperty; | ||
| import java.util.HashMap; | ||
| import java.util.Map; | ||
| import lombok.AllArgsConstructor; | ||
| import lombok.Data; | ||
| import lombok.NoArgsConstructor; | ||
| import org.springframework.boot.context.properties.ConfigurationProperties; | ||
|
|
||
| @ConfigurationProperties(prefix = "application.topics") | ||
| @Data | ||
| @NoArgsConstructor | ||
| @AllArgsConstructor | ||
| public class TopicProperties { | ||
|
|
||
| @JsonProperty("producers-cache-enabled") | ||
| private boolean producersCacheEnabled; | ||
| @JsonProperty("producers-cache-size") | ||
| private int producersCacheSize; | ||
|
|
||
| } |
104 changes: 104 additions & 0 deletions
104
...am-api-gateway/src/main/java/ai/langstream/apigateway/gateways/LRUTopicProducerCache.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,104 @@ | ||
| package ai.langstream.apigateway.gateways; | ||
|
|
||
| import ai.langstream.api.runner.code.Record; | ||
| import ai.langstream.api.runner.topics.TopicProducer; | ||
| import ai.langstream.api.runtime.Topic; | ||
| import com.google.common.cache.Cache; | ||
| import com.google.common.cache.CacheBuilder; | ||
| import com.google.common.cache.RemovalNotification; | ||
| import java.util.concurrent.CompletableFuture; | ||
| import java.util.concurrent.ExecutionException; | ||
| import java.util.concurrent.TimeUnit; | ||
| import java.util.function.Supplier; | ||
|
|
||
| public class LRUTopicProducerCache implements TopicProducerCache { | ||
|
|
||
|
|
||
| private static class SharedTopicProducer implements TopicProducer { | ||
| private final TopicProducer producer; | ||
| private volatile int referenceCount; | ||
| private volatile boolean cached = true; | ||
|
|
||
| public SharedTopicProducer(TopicProducer producer) { | ||
| this.producer = producer; | ||
| } | ||
|
|
||
| public void acquire() { | ||
| synchronized (this) { | ||
| referenceCount++; | ||
| } | ||
| } | ||
|
|
||
| public void removedFromCache() { | ||
| synchronized (this) { | ||
| cached = false; | ||
| if (referenceCount == 0) { | ||
| producer.close(); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public void start() { | ||
| producer.start(); | ||
| } | ||
|
|
||
| @Override | ||
| public void close() { | ||
| synchronized (this) { | ||
| referenceCount--; | ||
| if (referenceCount == 0 && !cached) { | ||
| producer.close(); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public CompletableFuture<?> write(Record record) { | ||
| return producer.write(record); | ||
| } | ||
|
|
||
| @Override | ||
| public Object getNativeProducer() { | ||
| return producer.getNativeProducer(); | ||
| } | ||
|
|
||
| @Override | ||
| public Object getInfo() { | ||
| return producer.getInfo(); | ||
| } | ||
|
|
||
| @Override | ||
| public long getTotalIn() { | ||
| return 0; | ||
| } | ||
| } | ||
|
|
||
|
|
||
| final Cache<Key, SharedTopicProducer> cache; | ||
|
|
||
| public LRUTopicProducerCache(int size) { | ||
| this.cache = CacheBuilder | ||
| .newBuilder() | ||
| .maximumSize(size) | ||
| .expireAfterWrite(10, TimeUnit.MINUTES) | ||
| .expireAfterAccess(10, TimeUnit.MINUTES) | ||
| .removalListener((RemovalNotification<Key, SharedTopicProducer> notification) -> { | ||
| SharedTopicProducer resource = notification.getValue(); | ||
| resource.removedFromCache(); | ||
| }) | ||
| .build(); | ||
| } | ||
|
|
||
| @Override | ||
| public synchronized TopicProducer getOrCreate(TopicProducerCache.Key key, Supplier<TopicProducer> topicProducerSupplier) { | ||
| try { | ||
| final SharedTopicProducer sharedTopicProducer = | ||
| cache.get(key, () -> new SharedTopicProducer(topicProducerSupplier.get())); | ||
| sharedTopicProducer.acquire(); | ||
| return sharedTopicProducer; | ||
| } catch (ExecutionException ex) { | ||
| throw new RuntimeException(ex); | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
10 changes: 10 additions & 0 deletions
10
...tream-api-gateway/src/main/java/ai/langstream/apigateway/gateways/TopicProducerCache.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,10 @@ | ||
| package ai.langstream.apigateway.gateways; | ||
|
|
||
| import ai.langstream.api.runner.topics.TopicProducer; | ||
| import java.util.function.Supplier; | ||
|
|
||
| public interface TopicProducerCache { | ||
| record Key(String tenant, String application, String gatewayId){} | ||
|
|
||
| TopicProducer getOrCreate(Key key, Supplier<TopicProducer> topicProducerSupplier); | ||
| } |
25 changes: 25 additions & 0 deletions
25
...pi-gateway/src/main/java/ai/langstream/apigateway/gateways/TopicProducerCacheFactory.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,25 @@ | ||
| package ai.langstream.apigateway.gateways; | ||
|
|
||
| import ai.langstream.api.runner.topics.TopicProducer; | ||
| import ai.langstream.api.storage.ApplicationStore; | ||
| import ai.langstream.apigateway.config.GatewayTestAuthenticationProperties; | ||
| import ai.langstream.apigateway.config.TopicProperties; | ||
| import java.util.function.Supplier; | ||
| import org.apache.commons.collections4.map.LRUMap; | ||
| import org.springframework.context.annotation.Bean; | ||
| import org.springframework.context.annotation.Configuration; | ||
|
|
||
| @Configuration | ||
| public class TopicProducerCacheFactory { | ||
|
|
||
| @Bean | ||
| public TopicProducerCache topicProducerCache( | ||
| TopicProperties topicProperties) { | ||
| if (topicProperties.isProducersCacheEnabled()) { | ||
| return new LRUTopicProducerCache(topicProperties.getProducersCacheSize()); | ||
| } else { | ||
| return (key, topicProducerSupplier) -> topicProducerSupplier.get(); | ||
| } | ||
|
|
||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.