1717import java .util .concurrent .atomic .AtomicBoolean ;
1818import java .util .concurrent .atomic .AtomicLong ;
1919import java .util .function .Consumer ;
20+ import java .util .regex .Pattern ;
2021
2122final class RedisMessagingDataAccess implements MessagingDataAccess {
2223
24+ private static final Pattern DESTINATION_PATTERN = Pattern .compile ("[A-Za-z0-9_.:-]{1,128}" );
25+
2326 private final JedisPool pool ;
2427 private final ExecutorService workers ;
2528 private final ILoggerAdapter logger ;
2629 private final MessageRegistry messageRegistry ;
2730 private final int maxSubscriptions ;
31+ private final int maxPayloadChars ;
2832 private final Map <String , ChannelSubscription > channelSubscriptions = new ConcurrentHashMap <>();
2933 private final Object subscriptionLock = new Object ();
3034 private final AtomicBoolean shuttingDown = new AtomicBoolean (false );
@@ -34,7 +38,8 @@ final class RedisMessagingDataAccess implements MessagingDataAccess {
3438 ExecutorService workers ,
3539 ILoggerAdapter logger ,
3640 MessageRegistry messageRegistry ,
37- int maxSubscriptions
41+ int maxSubscriptions ,
42+ int maxPayloadChars
3843 ) {
3944 this .pool = Objects .requireNonNull (pool , "Pool cannot be null" );
4045 this .workers = Objects .requireNonNull (workers , "Workers cannot be null" );
@@ -44,21 +49,29 @@ final class RedisMessagingDataAccess implements MessagingDataAccess {
4449 throw new IllegalArgumentException ("maxSubscriptions must be greater than zero" );
4550 }
4651 this .maxSubscriptions = maxSubscriptions ;
52+ if (maxPayloadChars < 1 ) {
53+ throw new IllegalArgumentException ("maxPayloadChars must be greater than zero" );
54+ }
55+ this .maxPayloadChars = maxPayloadChars ;
4756 }
4857
4958 @ Override
5059 public <T extends EventMessage > CompletableFuture <Void > publish (String dest , T msg ) {
51- Objects . requireNonNull ( dest , "Destination cannot be null" );
60+ String destination = validateDestination ( dest );
5261 Objects .requireNonNull (msg , "Message cannot be null" );
5362
5463 if (shuttingDown .get ()) {
5564 return CompletableFuture .failedFuture (new IllegalStateException ("Messaging provider is shutting down" ));
5665 }
5766
5867 String json = messageRegistry .toJson (msg );
68+ if (json .length () > maxPayloadChars ) {
69+ return CompletableFuture .failedFuture (new IllegalArgumentException (
70+ "Message payload exceeds maxPayloadChars (" + maxPayloadChars + ")" ));
71+ }
5972 return CompletableFuture .runAsync (() -> {
6073 try (Jedis j = pool .getResource ()) {
61- j .publish (dest , json );
74+ j .publish (destination , json );
6275 }
6376 }, workers );
6477 }
@@ -67,7 +80,7 @@ public <T extends EventMessage> CompletableFuture<Void> publish(String dest, T m
6780 public <T extends EventMessage > Subscription subscribe (
6881 String dest , Class <T > type , Consumer <T > handler
6982 ) {
70- Objects . requireNonNull ( dest , "Destination cannot be null" );
83+ String destination = validateDestination ( dest );
7184 Objects .requireNonNull (type , "Type cannot be null" );
7285 Objects .requireNonNull (handler , "Handler cannot be null" );
7386
@@ -78,13 +91,13 @@ public <T extends EventMessage> Subscription subscribe(
7891 ChannelSubscription channelSubscription ;
7992 boolean created = false ;
8093 synchronized (subscriptionLock ) {
81- channelSubscription = channelSubscriptions .get (dest );
94+ channelSubscription = channelSubscriptions .get (destination );
8295 if (channelSubscription == null ) {
8396 if (channelSubscriptions .size () >= maxSubscriptions ) {
8497 throw new IllegalStateException ("Maximum active Redis subscriptions reached (" + maxSubscriptions + ")" );
8598 }
86- channelSubscription = new ChannelSubscription (dest );
87- channelSubscriptions .put (dest , channelSubscription );
99+ channelSubscription = new ChannelSubscription (destination );
100+ channelSubscriptions .put (destination , channelSubscription );
88101 created = true ;
89102 }
90103 }
@@ -126,6 +139,15 @@ private ChannelSubscription(String destination) {
126139 this .pubSub = new JedisPubSub () {
127140 @ Override
128141 public void onMessage (String channel , String raw ) {
142+ if (raw == null || raw .isBlank ()) {
143+ logger .warn ("Received empty message while subscribing to channel " + channel );
144+ return ;
145+ }
146+ if (raw .length () > maxPayloadChars ) {
147+ logger .warn ("Dropped oversized message on channel " + channel + " (length="
148+ + raw .length () + ", maxPayloadChars=" + maxPayloadChars + ")" );
149+ return ;
150+ }
129151 for (HandlerRegistration <?> registration : handlers .values ()) {
130152 registration .dispatch (channel , raw );
131153 }
@@ -214,4 +236,12 @@ private void dispatch(String channel, String raw) {
214236 }
215237 }
216238 }
239+
240+ private static String validateDestination (String destination ) {
241+ Objects .requireNonNull (destination , "Destination cannot be null" );
242+ if (!DESTINATION_PATTERN .matcher (destination ).matches ()) {
243+ throw new IllegalArgumentException ("Destination contains unsupported characters." );
244+ }
245+ return destination ;
246+ }
217247}
0 commit comments