message expiration not work

[CrazyMan2]

import {
Connection,
Queue,
Exchange
} from 'react-native-rabbitmq';

let connection:Connection=null;
let queue:Queue;
let exchange:Exchange;
let mExchangeName="weixinFace";
let mCallFun=null;//回调函数

export function setCallFun(callFun) {
mCallFun=callFun;
}
export function connectRabbitMq(channel) {
const config = {
host: 'xxxxxx',
port: 5672,
username: 'xxxx', //your rabbitmq management username
password: 'xxxx', //your rabbitmq management password
virtualhost: '/',
};

connection = new Connection(config);
connection.connect();
let connected = false;

connection.on('connected', (event) => {
    console.log("connected");
    queue = new Queue(connection, {
        name: channel,
        passive: false,
        durable: true,
        exclusive: false,
        // consumer_arguments: { 'x-message-ttl': 1000 },
        // Arguments:{'x-message-ttl': 1000 },
        // ttl:1000,
        autoDelete: false,
    });
    exchange = new Exchange(connection, {
        name: mExchangeName,
        type: 'direct',
        durable:true,
        autoDelete: false,
        internal: false,
    });
    queue.bind(exchange, channel);
    queue.on('message', (data) => {
        console.log(data);
        if(mCallFun)mCallFun(data);
    });
});
connection.on('error', event => {
    connected = false;
    console.log(connection);
    console.log(event);
    console.log("you are not connected");
});

}
export function closeRabbitMq() {
// connection.close();
}
export function sendMessage(message,channel) {
let properties ={expiration:1000};
exchange.publish(message,channel,properties);
}

queue ttl and exchange properties expiration not work ,please help me

[CrazyMan2]

queue = new Queue(connection, {
name: channel,
passive: false,
durable: true,
exclusive: false,
consumer_arguments: { 'x-message-ttl': 1000 },
// Arguments:{'x-message-ttl': 1000 },
// ttl:1000,
autoDelete: false,
});

x-message-ttl not set success

[mrozkosz]

I have the same problem. Someone knows how to solve it?

[timhonders]

IOS or Android?

[timhonders]

Can u try this

queue = new Queue(connection, {
name: channel,
passive: false,
durable: true,
exclusive: false,
consumer_arguments: { },
autoDelete: false,
}, {
'x-message-ttl': 1000
});

[mrozkosz]

I have this problem in both platforms. On Android I chanced a variable in file RabbitMqQueue.java
before
this.channel.queueDeclare(this.name, this.durable, this.exclusive, this.autodelete, args);
now (it work)
this.channel.queueDeclare(this.name, this.durable, this.exclusive, this.autodelete, consumer_args);
Still I don't know how to solve this problem on iOS .

[timhonders]

ok that's not the way to do it, i will take al look at this

[timhonders]

This example works on android:

this.queue = new Queue(this.connection, {
name: 'test5.testqueue',
passive: false,
durable: true,
exclusive: false,
consumer_arguments: {
'x-priority': this.priority,
},
buffer_delay: 600,
autoBufferAck: true
}, {
'x-queue-mode': 'lazy',
'x-message-ttl': 1000,
'x-expires': 1800000
});

[mrozkosz]

Is there any solution for IOS?

[Ro392]

Hello friend, do you know how to solve this error on IOS? It is very important!

[Ro392]

IOS - sends arguments, but the queue created doesn't have them.

[timhonders]

Can u show youre code

[Ro392]

[foo-io]

Hi! Replace this code in node_modules/react-native-rabbit/ios/RCTReactNativeRabbitMq/RabbitMqQueue.m

#import "RabbitMqQueue.h"

@interface RabbitMqQueue ()
    @property (nonatomic, readwrite) NSString *name;
    @property (nonatomic, readwrite) NSDictionary *config;
    @property (nonatomic, readwrite) RMQQueue *queue; 
    @property (nonatomic, readwrite) id<RMQChannel> channel;
    @property (nonatomic, readwrite) RMQQueueDeclareOptions options;
    @property (nonatomic, readwrite) RCTBridge *bridge;
    @property (nonatomic, readwrite) RMQConsumer *consumer;
@end

@implementation RabbitMqQueue

RCT_EXPORT_MODULE();

-(id)initWithConfig:(NSDictionary *)config channel:(id<RMQChannel>)channel {
    if (self = [super init]) {

        self.config = config;
        self.channel = channel;
        self.name = [config objectForKey:@"name"];

        self.options = RMQQueueDeclareNoOptions;
    
        if ([config objectForKey:@"passive"] != nil && [[config objectForKey:@"passive"] boolValue]){
            self.options = self.options | RMQQueueDeclarePassive;
        }

        if ([config objectForKey:@"durable"] != nil && [[config objectForKey:@"durable"] boolValue]){
            self.options = self.options | RMQQueueDeclareDurable;
        }

        if ([config objectForKey:@"exclusive"] != nil && [[config objectForKey:@"exclusive"] boolValue]){
            self.options = self.options | RMQQueueDeclareExclusive;
        }

        if ([config objectForKey:@"autoDelete"] != nil && [[config objectForKey:@"autoDelete"] boolValue]){
            self.options = self.options | RMQExchangeDeclareAutoDelete;
        }

        if ([config objectForKey:@"NoWait"] != nil && [[config objectForKey:@"NoWait"] boolValue]){
            self.options = self.options | RMQQueueDeclareNoWait;
        }

        NSMutableDictionary *tmp_arguments = [[NSMutableDictionary alloc]init];
        if ([config objectForKey:@"consumer_arguments"] != nil){

            NSDictionary *consumer_arguments = [config objectForKey:@"consumer_arguments"];

            if ([consumer_arguments objectForKey:@"x-message-ttl"] != nil){
                NSNumber *ttl = [consumer_arguments objectForKey:@"x-message-ttl"];
                [tmp_arguments setObject:[[RMQSignedShort alloc] init:[ttl integerValue]] forKey:@"x-message-ttl"];
            }
            
            if ([consumer_arguments objectForKey:@"x-expires"] != nil){
                NSNumber *expires = [consumer_arguments objectForKey:@"x-expires"];
                [tmp_arguments setObject:[[RMQSignedLong alloc] init:[expires intValue]] forKey:@"x-expires"];
            }

            if ([consumer_arguments objectForKey:@"x-max-length"] != nil){
                NSNumber *maxLength = [consumer_arguments objectForKey:@"x-max-length"];
                [tmp_arguments setObject:[[RMQSignedLong alloc] init:[maxLength intValue]] forKey:@"x-max-length"];
            }

            if ([consumer_arguments objectForKey:@"x-priority"] != nil){
                NSNumber *xpriority = [consumer_arguments objectForKey:@"x-priority"];
                [tmp_arguments setObject:[[RMQSignedShort alloc] init:[xpriority integerValue]] forKey:@"x-priority"];
            }

            if ([consumer_arguments objectForKey:@"x-single-active-consumer"] != nil){
                BOOL active = [[consumer_arguments valueForKey:@"x-single-active-consumer"] boolValue];
                [tmp_arguments setObject:[[RMQBoolean alloc] init:active] forKey:@"x-single-active-consumer"];
            }

        }

        RMQBasicConsumeOptions consumer_options = RMQBasicConsumeNoOptions;

        RMQTable *arguments = [[RMQTable alloc] init:tmp_arguments];

        self.queue = [self.channel queue:self.name options:self.options arguments:tmp_arguments];

        self.consumer = [self.queue subscribe:consumer_options
                    arguments:arguments
                    handler:^(RMQMessage * _Nonnull message) {

            NSString *body = [[NSString alloc] initWithData:message.body encoding:NSUTF8StringEncoding];

            //[self.channel ack:message.deliveryTag];

            [EventEmitter emitEventWithName:@"RabbitMqQueueEvent" 
                body:@{
                    @"name": @"message", 
                    @"queue_name": self.name, 
                    @"message": body, 
                    @"routingKey": message.routingKey, // Will be deprecated
                    @"routing_key": message.routingKey, 
                    @"exchange": message.exchangeName,
                    @"consumer_tag": message.consumerTag, 
                    @"delivery_tag": message.deliveryTag
                }
            ];
    
        }];

    }
    return self;
}

-(void) bind:(RMQExchange *)exchange routing_key:(NSString *)routing_key {
    if ([routing_key length] == 0){
        [self.queue bind:exchange];
    }else{
        [self.queue bind:exchange routingKey:routing_key];
    }
}

-(void) unbind:(RMQExchange *)exchange routing_key:(NSString *)routing_key {
    if ([routing_key length] == 0){
        [self.queue unbind:exchange];
    }else{
        [self.queue unbind:exchange routingKey:routing_key];
    }
}

-(void) delete {
    [self.queue delete:self.options];
}

-(void) ack:(NSNumber *)deliveryTag {
    [self.channel ack:deliveryTag];
}

-(void) cancelConsumer {
    [self.consumer cancel];
}

@end

and example for use:

let queue = new Queue(connection, {
          name: 'queueName',
          passive: false,
          durable: true,
          exclusive: false,
          consumer_arguments: {
            'x-expires': 1800000,
            'x-max-length': 1,
          },
        });