Skip to content
GitLab
Menu
Projects
Groups
Snippets
Loading...
Help
Help
Support
Community forum
Keyboard shortcuts
?
Submit feedback
Contribute to GitLab
Sign in / Register
Toggle navigation
Menu
Open sidebar
ruanhaishen
redis
Commits
4774d616
Commit
4774d616
authored
Jun 12, 2018
by
antirez
Browse files
Streams: improve type correctness in t_stream.c.
parent
bcc42028
Changes
1
Show whitespace changes
Inline
Side-by-side
src/t_stream.c
View file @
4774d616
...
@@ -172,7 +172,7 @@ int streamCompareID(streamID *a, streamID *b) {
...
@@ -172,7 +172,7 @@ int streamCompareID(streamID *a, streamID *b) {
* if the ID was generated by the function. However the function may return
* if the ID was generated by the function. However the function may return
* C_ERR if an ID was given via 'use_id', but adding it failed since the
* C_ERR if an ID was given via 'use_id', but adding it failed since the
* current top ID is greater or equal. */
* current top ID is greater or equal. */
int
streamAppendItem
(
stream
*
s
,
robj
**
argv
,
int
numfields
,
streamID
*
added_id
,
streamID
*
use_id
)
{
int
streamAppendItem
(
stream
*
s
,
robj
**
argv
,
int
64_t
numfields
,
streamID
*
added_id
,
streamID
*
use_id
)
{
/* If an ID was given, check that it's greater than the last entry ID
/* If an ID was given, check that it's greater than the last entry ID
* or return an error. */
* or return an error. */
if
(
use_id
&&
streamCompareID
(
use_id
,
&
s
->
last_id
)
<=
0
)
return
C_ERR
;
if
(
use_id
&&
streamCompareID
(
use_id
,
&
s
->
last_id
)
<=
0
)
return
C_ERR
;
...
@@ -260,7 +260,7 @@ int streamAppendItem(stream *s, robj **argv, int numfields, streamID *added_id,
...
@@ -260,7 +260,7 @@ int streamAppendItem(stream *s, robj **argv, int numfields, streamID *added_id,
lp
=
lpAppendInteger
(
lp
,
1
);
/* One item, the one we are adding. */
lp
=
lpAppendInteger
(
lp
,
1
);
/* One item, the one we are adding. */
lp
=
lpAppendInteger
(
lp
,
0
);
/* Zero deleted so far. */
lp
=
lpAppendInteger
(
lp
,
0
);
/* Zero deleted so far. */
lp
=
lpAppendInteger
(
lp
,
numfields
);
lp
=
lpAppendInteger
(
lp
,
numfields
);
for
(
int
i
=
0
;
i
<
numfields
;
i
++
)
{
for
(
int
64_t
i
=
0
;
i
<
numfields
;
i
++
)
{
sds
field
=
argv
[
i
*
2
]
->
ptr
;
sds
field
=
argv
[
i
*
2
]
->
ptr
;
lp
=
lpAppend
(
lp
,(
unsigned
char
*
)
field
,
sdslen
(
field
));
lp
=
lpAppend
(
lp
,(
unsigned
char
*
)
field
,
sdslen
(
field
));
}
}
...
@@ -285,10 +285,10 @@ int streamAppendItem(stream *s, robj **argv, int numfields, streamID *added_id,
...
@@ -285,10 +285,10 @@ int streamAppendItem(stream *s, robj **argv, int numfields, streamID *added_id,
/* Check if the entry we are adding, have the same fields
/* Check if the entry we are adding, have the same fields
* as the master entry. */
* as the master entry. */
int
master_fields_count
=
lpGetInteger
(
lp_ele
);
int
64_t
master_fields_count
=
lpGetInteger
(
lp_ele
);
lp_ele
=
lpNext
(
lp
,
lp_ele
);
lp_ele
=
lpNext
(
lp
,
lp_ele
);
if
(
numfields
==
master_fields_count
)
{
if
(
numfields
==
master_fields_count
)
{
int
i
;
int
64_t
i
;
for
(
i
=
0
;
i
<
master_fields_count
;
i
++
)
{
for
(
i
=
0
;
i
<
master_fields_count
;
i
++
)
{
sds
field
=
argv
[
i
*
2
]
->
ptr
;
sds
field
=
argv
[
i
*
2
]
->
ptr
;
int64_t
e_len
;
int64_t
e_len
;
...
@@ -332,14 +332,14 @@ int streamAppendItem(stream *s, robj **argv, int numfields, streamID *added_id,
...
@@ -332,14 +332,14 @@ int streamAppendItem(stream *s, robj **argv, int numfields, streamID *added_id,
lp
=
lpAppendInteger
(
lp
,
id
.
seq
-
master_id
.
seq
);
lp
=
lpAppendInteger
(
lp
,
id
.
seq
-
master_id
.
seq
);
if
(
!
(
flags
&
STREAM_ITEM_FLAG_SAMEFIELDS
))
if
(
!
(
flags
&
STREAM_ITEM_FLAG_SAMEFIELDS
))
lp
=
lpAppendInteger
(
lp
,
numfields
);
lp
=
lpAppendInteger
(
lp
,
numfields
);
for
(
int
i
=
0
;
i
<
numfields
;
i
++
)
{
for
(
int
64_t
i
=
0
;
i
<
numfields
;
i
++
)
{
sds
field
=
argv
[
i
*
2
]
->
ptr
,
value
=
argv
[
i
*
2
+
1
]
->
ptr
;
sds
field
=
argv
[
i
*
2
]
->
ptr
,
value
=
argv
[
i
*
2
+
1
]
->
ptr
;
if
(
!
(
flags
&
STREAM_ITEM_FLAG_SAMEFIELDS
))
if
(
!
(
flags
&
STREAM_ITEM_FLAG_SAMEFIELDS
))
lp
=
lpAppend
(
lp
,(
unsigned
char
*
)
field
,
sdslen
(
field
));
lp
=
lpAppend
(
lp
,(
unsigned
char
*
)
field
,
sdslen
(
field
));
lp
=
lpAppend
(
lp
,(
unsigned
char
*
)
value
,
sdslen
(
value
));
lp
=
lpAppend
(
lp
,(
unsigned
char
*
)
value
,
sdslen
(
value
));
}
}
/* Compute and store the lp-count field. */
/* Compute and store the lp-count field. */
int
lp_count
=
numfields
;
int
64_t
lp_count
=
numfields
;
lp_count
+=
3
;
/* Add the 3 fixed fields flags + ms-diff + seq-diff. */
lp_count
+=
3
;
/* Add the 3 fixed fields flags + ms-diff + seq-diff. */
if
(
!
(
flags
&
STREAM_ITEM_FLAG_SAMEFIELDS
))
{
if
(
!
(
flags
&
STREAM_ITEM_FLAG_SAMEFIELDS
))
{
/* If the item is not compressed, it also has the fields other than
/* If the item is not compressed, it also has the fields other than
...
@@ -579,7 +579,7 @@ int streamIteratorGetID(streamIterator *si, streamID *id, int64_t *numfields) {
...
@@ -579,7 +579,7 @@ int streamIteratorGetID(streamIterator *si, streamID *id, int64_t *numfields) {
/* If we are going backward, read the number of elements this
/* If we are going backward, read the number of elements this
* entry is composed of, and jump backward N times to seek
* entry is composed of, and jump backward N times to seek
* its start. */
* its start. */
int
lp_count
=
lpGetInteger
(
si
->
lp_ele
);
int
64_t
lp_count
=
lpGetInteger
(
si
->
lp_ele
);
if
(
lp_count
==
0
)
{
/* We reached the master entry. */
if
(
lp_count
==
0
)
{
/* We reached the master entry. */
si
->
lp
=
NULL
;
si
->
lp
=
NULL
;
si
->
lp_ele
=
NULL
;
si
->
lp_ele
=
NULL
;
...
@@ -642,12 +642,12 @@ int streamIteratorGetID(streamIterator *si, streamID *id, int64_t *numfields) {
...
@@ -642,12 +642,12 @@ int streamIteratorGetID(streamIterator *si, streamID *id, int64_t *numfields) {
* forward, or seek the previous entry if we are going
* forward, or seek the previous entry if we are going
* backward. */
* backward. */
if
(
!
si
->
rev
)
{
if
(
!
si
->
rev
)
{
int
to_discard
=
(
flags
&
STREAM_ITEM_FLAG_SAMEFIELDS
)
?
int
64_t
to_discard
=
(
flags
&
STREAM_ITEM_FLAG_SAMEFIELDS
)
?
*
numfields
:
*
numfields
*
2
;
*
numfields
:
*
numfields
*
2
;
for
(
int64_t
i
=
0
;
i
<
to_discard
;
i
++
)
for
(
int64_t
i
=
0
;
i
<
to_discard
;
i
++
)
si
->
lp_ele
=
lpNext
(
si
->
lp
,
si
->
lp_ele
);
si
->
lp_ele
=
lpNext
(
si
->
lp
,
si
->
lp_ele
);
}
else
{
}
else
{
int
prev_times
=
4
;
/* flag + id ms + id seq + one more to
int
64_t
prev_times
=
4
;
/* flag + id ms + id seq + one more to
go back to the previous entry "count"
go back to the previous entry "count"
field. */
field. */
/* If the entry was not flagged SAMEFIELD we also read the
/* If the entry was not flagged SAMEFIELD we also read the
...
...
Write
Preview
Markdown
is supported
0%
Try again
or
attach a new file
.
Attach a file
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment