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
574017b7
You need to sign in or sign up before continuing.
Commit
574017b7
authored
Jun 12, 2018
by
antirez
Browse files
Streams: improve type correctness in t_stream.c.
parent
cd2b5a79
Changes
1
Hide whitespace changes
Inline
Side-by-side
src/t_stream.c
View file @
574017b7
...
@@ -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,14 +642,14 @@ int streamIteratorGetID(streamIterator *si, streamID *id, int64_t *numfields) {
...
@@ -642,14 +642,14 @@ 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
* number of fields, so go back one more. */
* number of fields, so go back one more. */
if
(
!
(
flags
&
STREAM_ITEM_FLAG_SAMEFIELDS
))
prev_times
++
;
if
(
!
(
flags
&
STREAM_ITEM_FLAG_SAMEFIELDS
))
prev_times
++
;
...
...
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